Refactor/338 : Rate Limiter 토큰 대기방식 개선 (#422)

* refactor : bucket 캡슐화

* refactor : 변경된 메서드 reader에 적용

* refactor : refillGreedy로 변경함에 따른 설정 변경

* refactor : reader에서 ACK 제거

* feat : RedisStreams와 상호작용하는 부분 모듈화

* refactor : ACK 시점 변경 및 RedisStreams 사용 부분 리팩토링

* feat : writer와 reader에 적용

* feat : 문제 출제 실패 시, MQ 데이터 삭제 추가

* feat : 문제 출제 실패로 인한 메일 발송 실패 상태 추가

* refactor : RateLimiter 토큰 대기방식 개선

* feat : 누락된 코드 추가

Author: crocusia

cs25-batch/src/main/java/com/example/cs25batch/batch/component/reader/RedisStreamReader.java

@@ -2,22 +2,14 @@
 
 import com.example.cs25batch.adapter.RedisStreamsClient;
 import com.example.cs25batch.sender.context.MailSenderContext;
-import io.github.bucket4j.Bucket;
 import java.time.Duration;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.batch.item.ItemReader;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.data.redis.connection.stream.Consumer;
 import org.springframework.data.redis.connection.stream.MapRecord;
-import org.springframework.data.redis.connection.stream.ReadOffset;
-import org.springframework.data.redis.connection.stream.StreamOffset;
-import org.springframework.data.redis.connection.stream.StreamReadOptions;
-import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 
 @Slf4j
@@ -35,9 +27,12 @@ public class RedisStreamReader implements ItemReader<Map<String, String>> {
     public Map<String, String> read() throws InterruptedException {
         //long start = System.currentTimeMillis();
 
+/*
         while (!mailSenderContext.tryConsume(strategyKey, 1L)) {
             Thread.sleep(200); //토큰을 얻을 때까지 간격을 두고 재시도
         }
+*/
+        mailSenderContext.acquirePermitOrWait(strategyKey);
 
         MapRecord<String, Object, Object> msg = redisClient.readWithConsumerGroup(Duration.ofMillis(500));
         //redisTemplate.opsForStream().acknowledge(STREAM, GROUP, msg.getId());

cs25-batch/src/main/java/com/example/cs25batch/sender/JavaMailSenderStrategy.java

@@ -4,6 +4,10 @@
 import com.example.cs25batch.batch.service.JavaMailService;
 import io.github.bucket4j.Bandwidth;
 import io.github.bucket4j.Bucket;
+import io.github.bucket4j.ConsumptionProbe;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.LockSupport;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Component;
 
@@ -17,7 +21,7 @@ public class JavaMailSenderStrategy implements MailSenderStrategy{
             .addLimit(
                     Bandwidth.builder()
                             .capacity(4)
-                            .refillGreedy(4, Duration.ofMillis(1000))
+                            .refillGreedy(2, Duration.ofMillis(500))
                             .build()
             )
             .build();
@@ -31,4 +35,18 @@ public void sendQuizMail(MailDto mailDto) {
     public boolean tryConsume(Long num){
         return bucket.tryConsume(num);
     }
+
+    @Override
+    public void acquirePermitOrWait(){
+        while (true) {
+            ConsumptionProbe probe = bucket.tryConsumeAndReturnRemaining(1);
+            if (probe.isConsumed()) return;
+
+            long nanos = probe.getNanosToWaitForRefill();
+            long jitter = TimeUnit.MILLISECONDS.toNanos(
+                ThreadLocalRandom.current().nextInt(0, 50)
+            );
+            LockSupport.parkNanos(Math.min(nanos + jitter, TimeUnit.SECONDS.toNanos(1)));
+        }
+    }
 }

cs25-batch/src/main/java/com/example/cs25batch/sender/MailSenderStrategy.java

@@ -1,9 +1,12 @@
 package com.example.cs25batch.sender;
 
 import com.example.cs25batch.batch.dto.MailDto;
+import io.github.bucket4j.Bucket;
 
 public interface MailSenderStrategy {
     void sendQuizMail(MailDto mailDto);
 
     boolean tryConsume(Long num);
+
+    void acquirePermitOrWait();
 }

cs25-batch/src/main/java/com/example/cs25batch/sender/SesMailSenderStrategy.java

@@ -4,6 +4,10 @@
 import com.example.cs25batch.batch.service.SesMailService;
 import io.github.bucket4j.Bandwidth;
 import io.github.bucket4j.Bucket;
+import io.github.bucket4j.ConsumptionProbe;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.LockSupport;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Component;
 
@@ -32,4 +36,18 @@ public void sendQuizMail(MailDto mailDto) {
     public boolean tryConsume(Long num){
         return bucket.tryConsume(num);
     }
+
+    @Override
+    public void acquirePermitOrWait(){
+        while (true) {
+            ConsumptionProbe probe = bucket.tryConsumeAndReturnRemaining(1);
+            if (probe.isConsumed()) return;
+
+            long nanos = probe.getNanosToWaitForRefill();
+            long jitter = TimeUnit.MILLISECONDS.toNanos(
+                ThreadLocalRandom.current().nextInt(0, 50)
+            );
+            LockSupport.parkNanos(Math.min(nanos + jitter, TimeUnit.SECONDS.toNanos(1)));
+        }
+    }
 }

cs25-batch/src/main/java/com/example/cs25batch/sender/context/MailSenderContext.java

@@ -30,4 +30,8 @@ private MailSenderStrategy getValidStrategy(String strategyKey) {
         return strategy;
     }
 
+    public void acquirePermitOrWait(String strategyKey) {
+        MailSenderStrategy strategy = getValidStrategy(strategyKey);
+        strategy.acquirePermitOrWait();
+    }
 }