diff --git a/build.gradle b/build.gradle index 8ee9733..64ac6a0 100644 --- a/build.gradle +++ b/build.gradle @@ -40,6 +40,8 @@ dependencies { testAnnotationProcessor 'org.projectlombok:lombok' implementation 'org.springframework.boot:spring-boot-starter-amqp' + + implementation 'org.springframework.boot:spring-boot-starter-data-redis' } tasks.named('test') { diff --git a/docker-compose.yml b/docker-compose.yml index 7332929..3d5492c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -24,6 +24,8 @@ services: MOCK_PROVIDER_FORCE_FAILURE: ${MOCK_PROVIDER_FORCE_FAILURE:-false} MOCK_PROVIDER_FAIL_FIRST_ATTEMPTS: ${MOCK_PROVIDER_FAIL_FIRST_ATTEMPTS:-0} OUTBOX_PUBLISH_INTERVAL_MS: ${OUTBOX_PUBLISH_INTERVAL_MS:-1000} + SPRING_DATA_REDIS_HOST: redis + SPRING_DATA_REDIS_PORT: 6379 healthcheck: test: @@ -41,6 +43,8 @@ services: condition: service_healthy rabbitmq: condition: service_healthy + redis: + condition: service_healthy prometheus: image: prom/prometheus:v3.13.0 @@ -129,6 +133,16 @@ services: timeout: 5s retries: 5 + redis: + image: redis:7.4 + ports: + - "6379:6379" + healthcheck: + test: [ "CMD", "redis-cli", "ping" ] + interval: 5s + timeout: 3s + retries: 10 + volumes: prometheus-data: grafana-data: diff --git a/src/main/java/com/backendsystemdesignlab/notification/messaging/DeliveryMessageHandler.java b/src/main/java/com/backendsystemdesignlab/notification/messaging/DeliveryMessageHandler.java index 8d9bdc9..c4280ae 100644 --- a/src/main/java/com/backendsystemdesignlab/notification/messaging/DeliveryMessageHandler.java +++ b/src/main/java/com/backendsystemdesignlab/notification/messaging/DeliveryMessageHandler.java @@ -5,6 +5,8 @@ import com.backendsystemdesignlab.notification.notification.provider.PushProvider; import com.backendsystemdesignlab.notification.notification.provider.SmsProvider; import com.backendsystemdesignlab.notification.notification.service.NotificationTransactionService; +import com.backendsystemdesignlab.notification.ratelimit.NotificationRateLimiter; +import com.backendsystemdesignlab.notification.ratelimit.ThrottlePublisher; import lombok.RequiredArgsConstructor; import org.springframework.stereotype.Service; @@ -21,10 +23,18 @@ public class DeliveryMessageHandler { private final DeliveryRetryPublisher retryPublisher; private final NotificationTransactionService transactionService; + private final NotificationRateLimiter rateLimiter; + private final ThrottlePublisher throttlePublisher; + public void handle(DeliveryMessage message) { if (transactionService.isAlreadyProcessed(message.deliveryId())) return; + if (!rateLimiter.tryAcquire(message.channel())) { + throttlePublisher.publish(message); + return; + } + boolean success = send(message); if (success) { diff --git a/src/main/java/com/backendsystemdesignlab/notification/messaging/RabbitMqConfig.java b/src/main/java/com/backendsystemdesignlab/notification/messaging/RabbitMqConfig.java index ebe7e02..efaf306 100644 --- a/src/main/java/com/backendsystemdesignlab/notification/messaging/RabbitMqConfig.java +++ b/src/main/java/com/backendsystemdesignlab/notification/messaging/RabbitMqConfig.java @@ -26,6 +26,14 @@ public class RabbitMqConfig { public static final String SMS_DLQ = "notification.sms.dlq"; public static final String EMAIL_DLQ = "notification.email.dlq"; + public static final String THROTTLE_EXCHANGE = "notification.throttle.exchange"; + public static final String PUSH_THROTTLE_QUEUE = "notification.push.throttle.queue"; + public static final String SMS_THROTTLE_QUEUE = "notification.sms.throttle.queue"; + public static final String EMAIL_THROTTLE_QUEUE = "notification.email.throttle.queue"; + public static final String PUSH_THROTTLE_ROUTING_KEY = "notification.push.throttle"; + public static final String SMS_THROTTLE_ROUTING_KEY = "notification.sms.throttle"; + public static final String EMAIL_THROTTLE_ROUTING_KEY = "notification.email.throttle"; + // Exchange: 메시지를 받아서 적절한 Queue로 전달하는 라우터 @Bean public DirectExchange notificationExchange() { @@ -43,6 +51,11 @@ public DirectExchange deadLetterExchange() { return new DirectExchange(DLX, true, false); } + @Bean + public DirectExchange throttleExchange() { + return new DirectExchange(THROTTLE_EXCHANGE, true, false); + } + @Bean public Queue pushQueue() { return new Queue(PUSH_QUEUE, true); @@ -109,6 +122,36 @@ public Queue emailDlq() { .build(); } + @Bean + public Queue pushThrottleQueue() { + return QueueBuilder + .durable(PUSH_THROTTLE_QUEUE) + .ttl(1000) + .deadLetterExchange(EXCHANGE) + .deadLetterRoutingKey(PUSH_ROUTING_KEY) + .build(); + } + + @Bean + public Queue smsThrottleQueue() { + return QueueBuilder + .durable(SMS_THROTTLE_QUEUE) + .ttl(1000) + .deadLetterExchange(EXCHANGE) + .deadLetterRoutingKey(SMS_ROUTING_KEY) + .build(); + } + + @Bean + public Queue emailThrottleQueue() { + return QueueBuilder + .durable(EMAIL_THROTTLE_QUEUE) + .ttl(1000) + .deadLetterExchange(EXCHANGE) + .deadLetterRoutingKey(EMAIL_ROUTING_KEY) + .build(); + } + @Bean public Binding pushBinding(DirectExchange notificationExchange, Queue pushQueue) { // pushQueue를 notificationExchange에 연결하고, notification.push라는 Routing Key로 연결해라 @@ -182,6 +225,30 @@ public Binding emailDlqBinding(DirectExchange deadLetterExchange, Queue emailDlq .with(EMAIL_ROUTING_KEY); } + @Bean + public Binding pushThrottleBinding(Queue pushThrottleQueue, DirectExchange throttleExchange) { + return BindingBuilder + .bind(pushThrottleQueue) + .to(throttleExchange) + .with(PUSH_THROTTLE_ROUTING_KEY); + } + + @Bean + public Binding smsThrottleBinding(Queue smsThrottleQueue, DirectExchange throttleExchange) { + return BindingBuilder + .bind(smsThrottleQueue) + .to(throttleExchange) + .with(SMS_THROTTLE_ROUTING_KEY); + } + + @Bean + public Binding emailThrottleBinding(Queue emailThrottleQueue, DirectExchange throttleExchange) { + return BindingBuilder + .bind(emailThrottleQueue) + .to(throttleExchange) + .with(EMAIL_THROTTLE_ROUTING_KEY); + } + @Bean public MessageConverter messageConverter() { return new JacksonJsonMessageConverter(); diff --git a/src/main/java/com/backendsystemdesignlab/notification/ratelimit/NotificationRateLimiter.java b/src/main/java/com/backendsystemdesignlab/notification/ratelimit/NotificationRateLimiter.java new file mode 100644 index 0000000..97ae8e7 --- /dev/null +++ b/src/main/java/com/backendsystemdesignlab/notification/ratelimit/NotificationRateLimiter.java @@ -0,0 +1,70 @@ +package com.backendsystemdesignlab.notification.ratelimit; + +import com.backendsystemdesignlab.notification.user.domain.NotificationChannel; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.core.io.ClassPathResource; +import org.springframework.data.redis.RedisConnectionFailureException; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.core.script.DefaultRedisScript; +import org.springframework.stereotype.Component; + +import java.time.Instant; +import java.util.List; + +@Component +@Slf4j +public class NotificationRateLimiter { + + private final StringRedisTemplate redisTemplate; + private final DefaultRedisScript script; + + private final int pushLimit; + private final int smsLimit; + private final int emailLimit; + + public NotificationRateLimiter( + StringRedisTemplate redisTemplate, + @Value("${notification.rate-limit.push:100}") int pushLimit, + @Value("${notification.rate-limit.sms:20}") int smsLimit, + @Value("${notification.rate-limit.email:50}") int emailLimit + ) { + this.redisTemplate = redisTemplate; + this.pushLimit = pushLimit; + this.smsLimit = smsLimit; + this.emailLimit = emailLimit; + + this.script = new DefaultRedisScript<>(); + this.script.setLocation(new ClassPathResource("scripts/rate-limit.lua")); + this.script.setResultType(Long.class); + } + + public boolean tryAcquire(NotificationChannel channel) { + + long second = Instant.now().getEpochSecond(); + + String key = "notification:rate-limit:" + channel.name() + ":" + second; + + try { + Long result = redisTemplate.execute( + script, + List.of(key), // KEYS[1] + String.valueOf(limit(channel)), // ARGV[1] + "2" // ARGV[2] + ); + + return result != null && result == 1L; + } catch (RedisConnectionFailureException e) { + log.warn("[RateLimit] Redis 사용 불가. Fail-open. channel={}", channel); + return true; + } + } + + private int limit(NotificationChannel channel) { + return switch (channel) { + case PUSH -> pushLimit; + case SMS -> smsLimit; + case EMAIL -> emailLimit; + }; + } +} diff --git a/src/main/java/com/backendsystemdesignlab/notification/ratelimit/ThrottlePublisher.java b/src/main/java/com/backendsystemdesignlab/notification/ratelimit/ThrottlePublisher.java new file mode 100644 index 0000000..7908d1d --- /dev/null +++ b/src/main/java/com/backendsystemdesignlab/notification/ratelimit/ThrottlePublisher.java @@ -0,0 +1,66 @@ +package com.backendsystemdesignlab.notification.ratelimit; + +import com.backendsystemdesignlab.notification.messaging.DeliveryMessage; +import com.backendsystemdesignlab.notification.messaging.RabbitMqConfig; +import com.backendsystemdesignlab.notification.outbox.OutboxEvent; +import com.backendsystemdesignlab.notification.user.domain.NotificationChannel; +import lombok.RequiredArgsConstructor; +import org.springframework.amqp.rabbit.connection.CorrelationData; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.stereotype.Component; + +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +@Component +@RequiredArgsConstructor +public class ThrottlePublisher { + + private static final long CONFIRM_TIMEOUT_SECONDS = 5; + + private final RabbitTemplate rabbitTemplate; + + public void publish(DeliveryMessage message) { + + try { + CorrelationData correlationData = new CorrelationData("throttle-" + message.deliveryId() + "-" + UUID.randomUUID()); + + rabbitTemplate.convertAndSend( + RabbitMqConfig.THROTTLE_EXCHANGE, + routingKey(message.channel()), + message, + correlationData + ); + + CorrelationData.Confirm confirm = correlationData.getFuture().get(CONFIRM_TIMEOUT_SECONDS, TimeUnit.SECONDS); // 5초 기다림 + + if (!confirm.ack()) { // RabbitMQ Broker가 잘 받았는지 (Publisher -> Broker) + throw new IllegalStateException("Throttle publish NACK. deliveryId=" + message.deliveryId() + ", reason=" + confirm.reason()); + } + + if (correlationData.getReturned() != null) { // Broker Exchange -> Queue로 라우팅됐는가? (라우팅 실패) 예: 잘못된 라우팅키 + throw new IllegalStateException("Throttle publish RETURN. deliveryId=" + message.deliveryId() + ", reason=" + correlationData.getReturned().getReplyText()); + } + } catch (TimeoutException e) { + throw new IllegalStateException("Throttle publisher confirm timeout. deliveryId=" + message.deliveryId(), e); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Thorttle publisher interrupted. deliveryId=" + message.deliveryId(), e); + } catch (Exception e) { + if (e instanceof IllegalStateException) { + throw (IllegalStateException) e; + } + + throw new IllegalStateException("Throttle publish failed. deliveryId=" + message.deliveryId(), e); + } + } + + private String routingKey(NotificationChannel channel) { + return switch (channel) { + case PUSH -> RabbitMqConfig.PUSH_THROTTLE_ROUTING_KEY; + case SMS -> RabbitMqConfig.SMS_THROTTLE_ROUTING_KEY; + case EMAIL -> RabbitMqConfig.EMAIL_THROTTLE_ROUTING_KEY; + }; + } +} diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index d74be54..84766c8 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -32,6 +32,11 @@ spring: template: mandatory: true + data: + redis: + host: ${SPRING_DATA_REDIS_HOST:localhost} + port: ${SPRING_DATA_REDIS_PORT:6379} + management: endpoints: web: @@ -60,4 +65,11 @@ mock: fail-first-attempts: ${MOCK_PROVIDER_FAIL_FIRST_ATTEMPTS:0} outbox: - publish-interval-ms: ${OUTBOX_PUBLISH_INTERVAL_MS:1000} \ No newline at end of file + publish-interval-ms: ${OUTBOX_PUBLISH_INTERVAL_MS:1000} + + +notification: + rate-limit: + push: ${PUSH_RATE_LIMIT:1000} + sms: ${SMS_RATE_LIMIT:5} + email: ${EMAIL_RATE_LIMIT:1000} \ No newline at end of file diff --git a/src/main/resources/scripts/rate-limit.lua b/src/main/resources/scripts/rate-limit.lua new file mode 100644 index 0000000..f61cd9a --- /dev/null +++ b/src/main/resources/scripts/rate-limit.lua @@ -0,0 +1,11 @@ +local current = redis.call('INCR', KEYS[1]) + +if current == 1 then + redis.call('EXPIRE', KEYS[1], ARGV[2]) +end + +if current <= tonumber(ARGV[1]) then + return 1 +end + +return 0 \ No newline at end of file