3가지 방식의 RateLimiter 로직 비교

RateLimiterTest.kt

package im.toss.home.batch

import io.github.resilience4j.kotlin.ratelimiter.executeSuspendFunction
import io.github.resilience4j.ratelimiter.RateLimiter
import io.github.resilience4j.ratelimiter.RateLimiterConfig
import io.github.resilience4j.ratelimiter.RateLimiterRegistry
import mu.KotlinLogging
import org.junit.jupiter.api.Test
import java.time.Duration
import java.time.LocalDateTime
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.flow.zip
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.supervisorScope

class RateLimiterTest {
    private val logger = KotlinLogging.logger { }

    @Test
    fun `두달간 문제 없던 방식`(): Unit = runBlocking {
        /**
         * 1초를 TPS로 나눠서 요청 사이에 쉬어야하는 시간을 균등하게 배분하는 방식이다
         * */
        fun tickerFlow(delayMillis: Long) = flow {
            while (true) {
                emit(Unit)
                delay(delayMillis)
            }
        }

        fun <T> Flow<T>.limitValuePerSeconds(valuePerSeconds: Int): Flow<T> {
            require(valuePerSeconds <= 1000) { "valuePerSeconds should be less then or equal to 1,000" }
            val delaySeconds: Double = 1.0 / valuePerSeconds.toDouble()
            val delayMillis: Long = (delaySeconds * 1000).toLong()
            return zip(tickerFlow(delayMillis)) { value, _ -> value }
        }

        // Arrange
        val expectedTps = 200

        // Act
        val callHistory = mutableListOf<LocalDateTime>()
        (0..1_000)
            .asFlow()
            .limitValuePerSeconds(expectedTps)
            .map {
                async {
                    callHistory.add(LocalDateTime.now())
                }
            }
            .toList()
            .awaitAll()

        // Assert
        assertCallHistory(callHistory, expectedTps)
    }

    @Test
    fun `RateLimiter 위험한 방식`(): Unit = runBlocking {
        /**
         * ## 문제 내용
         * RateLimiter에 TPS를 n으로 설정하면 전체 평균 TPS는 n으로 계산되지만
         * 특정 구간을 셈플링하면 2n의 TPS가 발견 됨
         * limitRefreshPeriod 사이즈로 설정된 Cycle의 초반에 요청이 몰릴 수 있기 때문임
         * 
         * limitRefreshPeriod 와 Cycle 개념 -> https://resilience4j.readme.io/docs/ratelimiter
        */
        runRateLimiterTest { tps ->
            limitRefreshPeriod(Duration.ofSeconds(1))
                .limitForPeriod(tps)
        }
    }

    @Test
    fun `RateLimiter 안전한 방식`(): Unit = runBlocking {
        /**
         * ## 해결 방법
         * limitRefreshPeriod 와 limitForPeriod 를 조정한다
         * 보통 200TPS를 설정할 때 경우 limitRefreshPeriod를 1초로 하고 limitForPeriod를 200으로 두는데,
         * Cycle 초반에 요청이 몰리는 것을 개선하기 위해 limitRefreshPeriod는 1/200으로 limitForPeriod는 1로 두면 된다 
         */
        runRateLimiterTest { tps ->
            limitRefreshPeriod(Duration.ofMillis(1000 / tps.toLong()))
                .limitForPeriod(1)
        }
    }

    private suspend fun runRateLimiterTest(
        limitTps: RateLimiterConfig.Builder.(Int) -> RateLimiterConfig.Builder,
    ): Unit = supervisorScope {
        // Arrange
        val expectedTps = 200
        val rateLimiter: RateLimiter = RateLimiterRegistry.ofDefaults()
            .rateLimiter(
                "test",
                RateLimiterConfig.custom()
                    .limitTps(expectedTps)
                    .timeoutDuration(Duration.ofMinutes(1000))
                    .build(),
            )

        // Act
        val callHistory = mutableListOf<LocalDateTime>()
        (0..1_000)
            .map {
                async {
                    rateLimiter.executeSuspendFunction {
                        callHistory.add(LocalDateTime.now())
                    }
                }
            }.awaitAll()

        // Assert
        assertCallHistory(callHistory, expectedTps)
    }

    private fun assertCallHistory(callHistory: List<LocalDateTime>, expectedTps: Int) {
        val tpsHistory: List<Int> = callHistory
            .mapIndexed { index, localDateTime ->
                callHistory.drop(index + 1).takeWhile { it <= localDateTime.plusSeconds(1) }
            }.map { it.size }.sortedDescending()
        logger.info { "TPS 평균 ${tpsHistory.average()}, TPS 기록: ${tpsHistory.take(100)}" }
        assert(tpsHistory.average() <= expectedTps)
        assert(tpsHistory.all { it <= expectedTps * 1.1 })
    }
}

val resilience4jVersion = "1.7.1"
implementation("io.github.resilience4j:resilience4j-ratelimiter:$resilience4jVersion")
implementation("io.github.resilience4j:resilience4j-reactor:$resilience4jVersion")
implementation("io.github.resilience4j:resilience4j-micrometer:$resilience4jVersion")
implementation("io.github.resilience4j:resilience4j-kotlin:$resilience4jVersion")

실행 결과

🟢 두달간 문제 없던 방식

12:11:58.865 [Test worker @coroutine#1003] INFO im.toss.home.batch.TestRateLimitter - TPS 평균 146.04395604395606, TPS 기록: [161, 161, 161, 161, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160, 160]

🔴 RateLimiter 위험한 방식

12:12:03.937 [Test worker @coroutine#2006] INFO im.toss.home.batch.TestRateLimitter - TPS 평균 217.03996003996005, TPS 기록: [398, 397, 396, 395, 394, 393, 392, 391, 390, 389, 388, 387, 386, 385, 384, 383, 382, 381, 380, 379, 378, 377, 376, 375, 374, 373, 372, 371, 370, 369, 368, 367, 366, 365, 364, 363, 362, 362, 361, 361, 361, 360, 360, 360, 359, 359, 359, 359, 358, 358, 358, 358, 357, 357, 357, 356, 356, 356, 355, 355, 355, 354, 354, 354, 353, 353, 353, 352, 352, 352, 351, 351, 351, 350, 350, 350, 349, 349, 349, 348, 348, 348, 347, 347, 347, 346, 346, 346, 345, 345, 345, 344, 344, 344, 343, 343, 343, 342, 342, 342]

🟢 RateLimiter 안전한 방식

12:11:52.105 [Test worker @coroutine#1] INFO im.toss.home.batch.TestRateLimitter - TPS 평균 179.92407592407594, TPS 기록: [217, 216, 215, 214, 213, 213, 212, 211, 210, 209, 208, 207, 206, 206, 205, 205, 204, 204, 203, 203, 202, 202, 202, 201, 201, 201, 201, 201, 201, 201, 201, 201, 201, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200]