feat: Redis 기반 알림 전송 Rate Limit 기능 추가

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') {

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:

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) {

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();

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<Long> 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;
+        };
+    }
+}

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;
+        };
+    }
+}

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}
+  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}

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