Spring Boot에 Redis와 연동하여 처리율 제한 장치(Rate Limiter) 적용하기(Spring Boot + Redis + Bucket4j)

Spring Boot에 Redis와 연동하여 처리율 제한 장치(Rate Limiter) 적용하기(Spring Boot + Redis + Bucket4j)

 

 

포스팅 동기

개인 프로젝트를 진행하면서 처리율 제한 장치를 적용해야하는 상황이 발생했습니다.

처리율 제한 장치를 구현하기 위해서는 크게 2가지 방법이 있는데,
Client와 Server 사이에 MiddleWare를 두어서 Gateway를 이용하는 방법과 Server에서 구현하는 방법이 있습니다.

 

저는 Gateway를 사용하기에는 규모가 작은 프로젝트라서 Server에 직접 구현하였습니다.^^

 

추가로 처리율 제한 장치에 여러 알고리즘이 존재하는데,
리소스를 고려하여 구현이 간편한 토큰 버킷 알고리즘을 사용했습니다.^^

 

이 모든 내용을 혼자만 알고있기에는 아쉬워서 공유도 하고 정리할 겸 포스팅을 하게되었습니다.😃

• 깃허브 링크

GitHub - ssosee/springboot-rate-limiter

 

 

처리율 제한 장치에 대해서 자세히 알아보고 싶은 분은 아래 링크를 참고 하세요.^^

• 2023.10.09 - [0 + 독서/가상 면접 사례로 배우는 대규모 설계 기초] - [가상 면접 사례로 배우는 대규모 설계 기초] 2. 처리율 제한 장치(Rate Limiter)

[가상 면접 사례로 배우는 대규모 설계 기초] 2. 처리율 제한 장치(Rate Limiter)

 

 

처리율 제한 장치?

먼저, 처리율 제한 장치에 대해서 간략하게 말씀드리겠습니다.^^
처리율 제한 장치는 클라이언트가 보내는 트래픽의 처리율을 제한하는 장치 입니다.

 

왜 제한을 할까요? 🤔
가장 큰 이유는 서버 자원 낭비를 막기 위해서 입니다.
낭비되지 않는 자원 만큼 비용도 절감 할 수 있겠죠? ㅎㅎ

 

처리율 제한 장치의 원리는 간단합니다.
클라이언트의 요청이 우리가 설정한 한도를 넘어가면 그 이후의 요청을 거부하면 됩니다!!

 

프로젝트 소개

프로젝트에 대해서 간단하게 소개하겠습니다.

 

참고
프로젝트에 대한 코드는 링크에 들어가면 있습니다.

 

애플리케이션 흐름

• AOP에 처리율 제한 장치를 적용 했습니다.
  • 물론, Filter 또는 Interceptor에 적용하셔도 됩니다. 👍
• 프로젝트 편의성을 위해 EmbeddedRedisServer를 사용했습니다.

 

처리율 제한 장치 정책 준수

 

 

처리율 제한 장치 정책 위반

 

 

핵심 코드

자! 코드 전체를 보기 전에 핵심 코드와 테스트 코드 를 살펴볼까요?
테스트 코드와 함께 보면 더 쉽게 이해하실 수 있을 거에요~!! 😆

 

RatePlan

• 처리율 제한 장치의 정책(plan)을 구현한 enum 클래스 입니다.
• enum을 사용하고 추상 메서드를 오버라이드 하여 다형성을 구현했습니다.
• TEST, LOCAL 2가지 정책이 존재 합니다.

@Getter
@RequiredArgsConstructor
public enum RatePlan {

    TEST("test") {
        @Override
        public Bandwidth getLimit() {
            return Bandwidth.classic(100, Refill.intervally(100, Duration.ofMinutes(1)));
        }
    },
    LOCAL("local") {
        @Override
        public Bandwidth getLimit() {
            return Bandwidth.classic(5, Refill.intervally(5, Duration.ofMinutes(1)));
        }
    };

    // 추상 메서드 선언
    public abstract Bandwidth getLimit();

    private final String planName;

    public static Bandwidth resolvePlan(String targetPlan) {
        if(targetPlan.equals(TEST.getPlanName())) return TEST.getLimit();
        else if(targetPlan.equals(LOCAL.getPlanName())) return LOCAL.getLimit();

        throw new RateLimiterException(RateLimiterException.NOT_FOUND);
    }
}

 

RatePlanTest

• 처리율 제한 장치의 정책에 해당하는 부분의 테스트 코드 입니다.
• RatePlan.resolvePlan(String plan) 메소드를 이용하면 알맞은 Bandwidth를 반환하는 것을 확인 할 수 있습니다.

class RatePlanTest {

    @Test
    @DisplayName("설정한 정책에 맞는 처리율 제한 장치 정책을 반환한다.")
    void resolvePlan() {
        // given
        String testPlan = "test";
        String localPlan = "local";

        // when
        Bandwidth testBandwidth = RatePlan.resolvePlan(testPlan);
        Bandwidth localBandwidth = RatePlan.resolvePlan(localPlan);

        // then
        assertAll(
                () -> assertThat(testBandwidth).isEqualTo(RatePlan.TEST.getLimit()),
                () -> assertThat(localBandwidth).isEqualTo(RatePlan.LOCAL.getLimit())
        );
    }

    @Test
    @DisplayName("설정한 정책이 처리율 제한 장치에 없으면 예외가 발생한다.")
    void resolvePlanException() {
        // given
        String myPlan = "myPlan";

        // when // then
        assertThatThrownBy(() -> RatePlan.resolvePlan(myPlan))
                .isInstanceOf(RateLimiterException.class)
                .hasMessage(RateLimiterException.NOT_FOUND);
    }
}

 

TokenBucketResolver

• 토큰 버킷 알고리즘을 사용하는 클래스 입니다.
• key값에 따라서 카운터를 계산합니다.
  • 카운터는 Redis에 보관 중 입니다.

@Component
@RequiredArgsConstructor
public class TokenBucketResolver {

    private final BucketConfiguration bucketConfiguration;
    private final LettuceBasedProxyManager lettuceBasedProxyManager;

    private Bucket bucket(String key) {
        return lettuceBasedProxyManager.builder()
                .build(key.getBytes(), bucketConfiguration);
    }

    public boolean checkBucketCounter(String key) {
        Bucket bucket = bucket(key);
        if(!bucket.tryConsume(1)) {
            throw new RateLimiterException(RateLimiterException.TOO_MANY_REQUEST);
        }

        return true;
    }

    public long getAvailableTokens(String key) {
        return bucket(key).getAvailableTokens();
    }
}

 

TokenBucketResolverTest

• 위에서 설정한 RatePlan에 맞는 정책을 적용한 토큰 버킷 알고리즘의 구현체를 테스트하는 코드 입니다.
• redisTemplate을 사용하여 테스트가 종료되면 초기화 해줍니다.
  • 처리율 제한 장치에서 사용하는 counter를 저장하는 저장소를 redis 사용
  • 초기화 하지 않으면, 테스트의 독립성을 보장할 수 없음

@SpringBootTest
class TokenBucketResolverTest {

    @Autowired
    TokenBucketResolver tokenBucketResolver;

    @Autowired
    RedisTemplate<?, ?> redisTemplate;

    @AfterEach
    void tearDown() {
        redisTemplate.getConnectionFactory()
                .getConnection()
                .flushAll();
    }

    @Test
    @DisplayName("처리율 제한 장치의 정책을 준수하면 true를 반환한다.")
    void checkBucketCounter() {
        // given
        String key = "1";

        // when
        boolean result = tokenBucketResolver.checkBucketCounter(key);

        // then
        assertTrue(result);
    }

    @Test
    @DisplayName("남은 토큰의 갯수를 반환한다.")
    void getAvailableTokens() {
        // given
        String key = "1";

        // when
        long availableTokens = tokenBucketResolver.getAvailableTokens(key);

        // then
        assertThat(availableTokens).isEqualTo(100);
    }

    @Test
    @DisplayName("처리율 제한 장치의 정책을 위반하면 예외가 발생한다.")
    void checkBucketCounterException() {
        // given
        String key = "1";

        // when
        for(int i = 0; i < 100; i++) {
            tokenBucketResolver.checkBucketCounter(key);
        }

        // then
        assertThatThrownBy(() -> tokenBucketResolver.checkBucketCounter(key))
                .isInstanceOf(RateLimiterException.class)
                .hasMessage(RateLimiterException.TOO_MANY_REQUEST);
    }

    @Test
    @DisplayName("key 값에 따라서 처리율 제한 장치의 정책이 적용된다.")
    void checkBucketCounterAppliedAccordingToKey() {
        // given
        String key1 = "1";
        String key2 = "2";

        // when
        for(int i = 0; i < 100; i++) {
            tokenBucketResolver.checkBucketCounter(key1);
        }
        boolean key2Result = tokenBucketResolver.checkBucketCounter(key2);

        // then
        assertAll(
                () -> assertThatThrownBy(() -> tokenBucketResolver.checkBucketCounter(key1))
                        .isInstanceOf(RateLimiterException.class)
                        .hasMessage(RateLimiterException.TOO_MANY_REQUEST),
                () -> assertTrue(key2Result)
        );
    }
}

 

MyController

• 컨트롤러 입니다.

@RestController
@RequiredArgsConstructor
public class MyController {

    private final TokenBucketResolver tokenBucketResolver;

    @GetMapping("/hello/{id}")
    public String hello(@PathVariable String id) {
        long availableTokens = tokenBucketResolver.getAvailableTokens(id);
        return "hello world!, counter="+availableTokens;
    }
}

 

MyControllerTest

• 컨트롤러 테스트 코드 입니다.

@SpringBootTest
@AutoConfigureMockMvc
class MyControllerTest {

    @Autowired
    MockMvc mockMvc;

    @Autowired
    RedisTemplate<?, ?> redisTemplate;

    @Autowired
    TokenBucketResolver tokenBucketResolver;

    @AfterEach
    void tearDown() {
        redisTemplate.getConnectionFactory().getConnection().flushAll();
    }

    @Test
    @DisplayName("hello world를 요청한다.")
    void hello() throws Exception {
        // given
        String key = "1";

        // when // then
        mockMvc.perform(get("/hello/{id}", key)
                        .contentType(MediaType.APPLICATION_JSON))
                .andDo(MockMvcResultHandlers.print())
                .andExpect(status().isOk())
                .andExpect(content().string("hello world!, counter="+tokenBucketResolver.getAvailableTokens(key)));
    }

    @Test
    @DisplayName("처리율 제한 장치의 정책을 위반하면 예외 응답을 준다.")
    void helloRateLimiterException() throws Exception {
        // given
        String key = "1";

        // when
        for(int i = 0; i < 100; i++) {
            mockMvc.perform(get("/hello/{id}", key)
                            .contentType(MediaType.APPLICATION_JSON))
                    .andDo(MockMvcResultHandlers.print())
                    .andExpect(status().isOk())
                    .andExpect(content().string("hello world!, counter="+tokenBucketResolver.getAvailableTokens(key)));
        }

        // then
        mockMvc.perform(get("/hello/{id}", key)
                    .contentType(MediaType.APPLICATION_JSON))
                .andDo(MockMvcResultHandlers.print())
                .andExpect(status().isTooManyRequests())
                .andExpect(content().string(RateLimiterException.TOO_MANY_REQUEST));
    }
}

 

 

프로젝트 전체 코드

bulid.gradle

• embedded-redis 라이브러리에 slf4j가 포함되어 있어 제외했습니다.
• bucket4j 라이브러리를 사용했습니다.

plugins {
    id 'java'
    id 'org.springframework.boot' version '2.7.16'
    id 'io.spring.dependency-management' version '1.0.15.RELEASE'
}

group = 'seaung'
version = '0.0.1-SNAPSHOT'

java {
    sourceCompatibility = '11'
}

configurations {
    compileOnly {
        extendsFrom annotationProcessor
    }
}

repositories {
    mavenCentral()
}

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-data-redis'
    implementation 'org.springframework.boot:spring-boot-starter-web'
    implementation 'org.springframework.boot:spring-boot-starter-aop'

    // https://mvnrepository.com/artifact/it.ozimov/embedded-redis
    implementation ('it.ozimov:embedded-redis:0.7.3') { exclude group: "org.slf4j", module: "slf4j-simple" }

    // rateLimiter
    implementation group: 'com.giffing.bucket4j.spring.boot.starter', name: 'bucket4j-spring-boot-starter', version: '0.6.0'

    // https://mvnrepository.com/artifact/com.github.vladimir-bukhtoyarov/bucket4j-redis
    implementation group: 'com.github.vladimir-bukhtoyarov', name: 'bucket4j-redis', version: '7.6.0'

    compileOnly 'org.projectlombok:lombok'
    annotationProcessor 'org.projectlombok:lombok'
    testImplementation 'org.springframework.boot:spring-boot-starter-test'
}

tasks.named('test') {
    useJUnitPlatform()
}

 

.yml

src > resources > application.yml

spring:
  profiles:
    default: local

 

src > resources > application-local.yml

bucket:
  plan: local

spring:
  config:
    activate:
      on-profile: local

  redis:
    host: localhost
    port: 6379

 

test > resources > application.yml

bucket:
  plan: test

spring:
  profiles:
    active: test

  redis:
    host: localhost
    port: 6379

 

EmbeddedRedisServerConfig

• EmbeddedRedis 설정
• EmbeddedRedis는 H2와 같은 내장 Redis 데몬
• EmbeddedRedis를 이용하여 테스트 코드를 수행하게 되면, 여러 스프링 테스트 컨텍스트가 실행되면서 EmbeddedRedis port 충돌
  • 따라서 다른 port를 사용하여 실행 할 수 있도록 한다.

@Slf4j
@Configuration
@Profile({"test", "local"})
public class EmbeddedRedisServerConfig {

    @Value("${spring.redis.port}")
    private int redisPort;
    private RedisServer redisServer;

    @PostConstruct
    public void runRedis() throws IOException {
        int port = isRedisServerRunning() ? findAvailablePort() : redisPort;
        redisServer = new RedisServer(port);
        redisServer.start();
        log.info("내장 레디스 서버 실행(port: {})", port);
    }

    @PreDestroy
    public void stopRedis() {
        if(!ObjectUtils.isEmpty(redisServer)) {
            redisServer.stop();
            log.info("내장 레디스 서버 종료");
        }
    }

    /**
     * 내장 레디스 서버가 현재 실행중인지 확인
     */
    private boolean isRedisServerRunning() throws IOException {
        return isRunning(executeGrepProcessCommand(redisPort));
    }

    /**
     * 내 PC에서 사용가능한 port 조회
     */
    public int findAvailablePort() throws IOException {
        for(int port = 10000; port <= 65535; port++) {
            Process process = executeGrepProcessCommand(port);
            if(!isRunning(process)) return port;
        }

        throw new IllegalArgumentException("10000 ~ 65535 에서 사용가능한 port를 찾을 수 없습니다.");
    }

    /**
     * 해당 port를 사용중인 process를 확인하는 sh 실행
     * 참고로 윈도우에서는 안됩니다. 맥/리눅스에서만 가능합니다.
     * 윈도우에서 동일하게 사용하시려면 exe 프로세스 찾는 코드를 작성해야합니다.
     */
    private Process executeGrepProcessCommand(int port) throws IOException {
        String command = String.format("netstat -nat | grep LISTEN|grep %d", port);
        String[] shell = {"/bin/sh", "-c", command};
        return Runtime.getRuntime().exec(shell);
    }

    /**
     * 해당 process가 실행 중인지 확인
     */
    private boolean isRunning(Process process) {
        String line;
        StringBuilder pidInfo = new StringBuilder();

        try(BufferedReader input = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
            while ((line = input.readLine()) != null) {
                pidInfo.append(line);
            }
        } catch (Exception e) {
            log.error("isRunning ", e);
        }

        return !StringUtils.isEmpty(pidInfo.toString());
    }
}

 

RedisConfig

• Spring Data Redis 설정 합니다.
• RedisConnectionFactory를 통해 내장 혹은 외부의 Redis와 연결 합니다.
• RedisTemplate을 통해 RedisConnection에서 넘겨준 byte 값을 객체 직렬화합니다.
• embeddedRedis가 먼저 실행되어야 하기 때문에 @DependsOn({"embeddedRedisServerConfig”})를 작성합니다.

@Slf4j
@Configuration
@DependsOn({"embeddedRedisServerConfig"}) // 빈 초기화 순서 지정
public class RedisConfig {

    @Value("${spring.redis.host}")
    private String redisHost;

    @Value("${spring.redis.port}")
    private int redisPort;

    /**
     * Multi-Thread 에서 Thread-Safe한 Redis 클라이언트로 netty에 의해 관리된다.
     * Sentinel, Cluster, Redis data model 같은 고급 기능들을 지원하며
     * 비동기 방식으로 요청하기에 TPS/CPU/Connection 개수와 응답속도 등 전 분야에서 Jedis 보다 뛰어나다.
     * 스프링 부트의 기본 의존성은 현재 Lettuce로 되어있다.
     */
    @Bean
    public RedisConnectionFactory redisConnectionFactory() {
        log.info("RedisConnectionFactory 초기화");
        return new LettuceConnectionFactory(redisHost, redisPort);
    }

    /**
     * Redis data access code를 간소화 하기 위해 제공되는 클래스이다.
     * 주어진 객체들을 자동으로 직렬화/역직렬화 하며 binary 데이터를 Redis에 저장한다.
     * 기본설정은 JdkSerializationRedisSerializer 이다.
     */
    @Bean
    public RedisTemplate<?, ?> redisTemplate() {
        log.info("RedisTemplate 초기화");
        RedisTemplate<byte[], byte[]> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory());
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());

        return redisTemplate;
    }
}

 

RateLimiterConfig

• 처리율 제한 장치를 설정 합니다.
• Redis와 연결하기 위한 RedisClient를 생성하고, 정책(RatePlan)을 설정 합니다.
• embeddedRedis가 먼저 실행되어야 하기 때문에 @DependsOn({"embeddedRedisServerConfig”})를 작성합니다.

@Slf4j
@Configuration
@DependsOn({"embeddedRedisServerConfig"}) // 빈 초기화 순서 지정
public class RateLimiterConfig {

    @Value("${spring.redis.host}")
    private String redisHost;

    @Value("${spring.redis.port}")
    private int redisPort;

    @Value("${bucket.plan}")
    private String bucketPlan;

    @Bean
    public RedisClient redisClient() {
        return RedisClient.create(RedisURI.builder()
                .withHost(redisHost)
                .withPort(redisPort)
                .build());
    }

    @Bean
    public LettuceBasedProxyManager lettuceBasedProxyManager() {
        return LettuceBasedProxyManager
                .builderFor(redisClient())
                .withExpirationStrategy(ExpirationAfterWriteStrategy
                        .basedOnTimeForRefillingBucketUpToMax(Duration.ofSeconds(10))
                )
                .build();
    }

    @Bean
    public BucketConfiguration bucketConfiguration() {
        log.info("처리율 제한 장치 정책={}", bucketPlan);
        return BucketConfiguration.builder()
                .addLimit(RatePlan.resolvePlan(bucketPlan))
                .build();
    }
}

 

RateLimiterException

• 처리율 제한 장치의 예외를 정의 합니다.

public class RateLimiterException extends IllegalArgumentException {

    public static final String TOO_MANY_REQUEST = "너무 많은 요청을 보냈습니다.";
    public static final String NOT_FOUND = "처리율 제한 장치 플랜을 찾을 수 없습니다.";

    public RateLimiterException() {
        super();
    }

    public RateLimiterException(String s) {
        super(s);
    }

    public RateLimiterException(String message, Throwable cause) {
        super(message, cause);
    }

    public RateLimiterException(Throwable cause) {
        super(cause);
    }
}

 

RatePlan

• 처리율 제한 장치의 정책(plan)을 구현한 enum 클래스 입니다.
• enum을 사용하고 추상 메서드를 오버라이드 하여 다형성을 구현했습니다.
• TEST, LOCAL 2가지 정책이 존재 합니다.

@Getter
@RequiredArgsConstructor
public enum RatePlan {

    TEST("test") {
        @Override
        public Bandwidth getLimit() {
            return Bandwidth.classic(100, Refill.intervally(100, Duration.ofMinutes(1)));
        }
    },
    LOCAL("local") {
        @Override
        public Bandwidth getLimit() {
            return Bandwidth.classic(5, Refill.intervally(5, Duration.ofMinutes(1)));
        }
    };

    public abstract Bandwidth getLimit();

    private final String planName;

    public static Bandwidth resolvePlan(String targetPlan) {
        if(targetPlan.equals(TEST.getPlanName())) return TEST.getLimit();
        else if(targetPlan.equals(LOCAL.getPlanName())) return LOCAL.getLimit();

        throw new RateLimiterException(RateLimiterException.NOT_FOUND);
    }
}

 

TokenBucketResolver

• 토큰 버킷 알고리즘을 사용하는 클래스 입니다.
• key값에 따라서 카운터를 계산합니다.
  • 카운터는 Redis에 보관 중 입니다.

@Component
@RequiredArgsConstructor
public class TokenBucketResolver {

    private final BucketConfiguration bucketConfiguration;
    private final LettuceBasedProxyManager lettuceBasedProxyManager;

    private Bucket bucket(String key) {
        return lettuceBasedProxyManager.builder()
                .build(key.getBytes(), bucketConfiguration);
    }

    public boolean checkBucketCounter(String key) {
        Bucket bucket = bucket(key);
        // 토큰 1개를 소비할 수 없으면
        if(!bucket.tryConsume(1)) {
            throw new RateLimiterException(RateLimiterException.TOO_MANY_REQUEST);
        }
        return true;
    }
}

 

AdviceController

• RateLimiterException.class 예외를 핸들링하는 클래스 입니다.
• [429] Too Many Requests 메시지를 응답으로 내려줍니다.

@Slf4j
@RestControllerAdvice
public class AdviceController {
    @ExceptionHandler(RateLimiterException.class)
    public ResponseEntity<String> rateLimiterException(RateLimiterException e) {
        log.error("예외 발생 ", e);
        return new ResponseEntity<>(e.getMessage(), HttpStatus.TOO_MANY_REQUESTS);
    }
}

 

MyController

• 컨트롤러 입니다.

@RestController
@RequiredArgsConstructor
public class MyController {

    private final TokenBucketResolver tokenBucketResolver;

    @GetMapping("/hello/{id}")
    public String hello(@PathVariable String id) {
        long availableTokens = tokenBucketResolver.getAvailableTokens(id);
        return "hello world!, counter="+availableTokens;
    }
}

 

AppPointCuts

• Controller가 포함된 클래스에 PointCut을 설정합니다.

public class AppPointCuts {

    @Pointcut("execution(* *..*Controller.*(..))")
    public void allController() {}
}

 

ControllerAspect

• AOP를 사용하여 처리율 제한 장치 적용 합니다.

@Slf4j
@Aspect
@Component
@RequiredArgsConstructor
public class ControllerAspect {
    private final TokenBucketResolver tokenBucketResolver;

    @Around("seaung.ratelimiter.aop.AppPointCuts.allController() && args(id)")
    public Object doController(ProceedingJoinPoint joinPoint, String id) throws Throwable {
        try {
            log.info("[doController] 컨트롤러 호출 전 {}", joinPoint.getClass());
            // 처리율 제한 장치 실행
            tokenBucketResolver.checkBucketCounter(id);
        } catch (Exception e) {
            log.error("[doController] 예외 발생", e);
            throw e;
        }

        return joinPoint.proceed();
    }
}

 

결과

참고

• Bucket4j 7.6.0 Reference
• Redis SpringBoot Data Redis 로컬/통합 테스트 환경 구축하기