구현

이 장에서는 토큰 버킷과 슬라이딩 창 카운터를 Python으로 구현하고, 토큰 버킷을 C++, Java, TypeScript로 옮깁니다. 시계를 바깥에서 넣어 테스트에서 시간을 직접 움직일 수 있게 합니다.

Python: 토큰 버킷과 슬라이딩 창 카운터

import threading
import time
from typing import Callable


class TokenBucket:
    def __init__(self, rate: float, capacity: float,
                 clock: Callable[[], float] = time.monotonic) -> None:
        if rate <= 0 or capacity < 1:
            raise ValueError("rate must be > 0 and capacity >= 1")
        self.rate, self.capacity, self.clock = rate, capacity, clock
        self.tokens = float(capacity)
        self.updated = clock()
        self.lock = threading.Lock()

    def try_acquire(self, cost: float = 1.0) -> tuple[bool, float]:
        """(허용 여부, 다시 시도까지 기다릴 초)"""
        with self.lock:
            now = self.clock()
            elapsed = max(0.0, now - self.updated)
            self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
            self.updated = now
            if self.tokens >= cost:
                self.tokens -= cost
                return True, 0.0
            return False, (cost - self.tokens) / self.rate


class SlidingWindowCounter:
    def __init__(self, limit: int, window: float,
                 clock: Callable[[], float] = time.time) -> None:
        self.limit, self.window, self.clock = limit, window, clock
        self.index = int(clock() // window)   # 현재 창 번호
        self.current = self.previous = 0
        self.lock = threading.Lock()

    def allow(self) -> bool:
        with self.lock:
            now = self.clock()
            index = int(now // self.window)
            if index != self.index:
                # 바로 다음 창이면 현재가 직전이 되고, 더 지났으면 0
                self.previous = self.current if index == self.index + 1 else 0
                self.current, self.index = 0, index
            weight = 1.0 - (now - index * self.window) / self.window
            if self.previous * weight + self.current < self.limit:
                self.current += 1
                return True
            return False

• clock: 기본값 time.monotonic은 NTP 보정으로 뒤로 갈 수 있는 벽시계와 달리 항상 앞으로만 갑니다. 창 카운터는 창 번호를 달력 시각과 맞추려고 벽시계를 씁니다.
• max(0.0, ...)는 시계가 뒤로 가도 토큰이 줄지 않게 하고, min(self.capacity, ...)는 오래 쉬어도 버스트가 용량을 넘지 않게 묶습니다.
• 두 번째 반환값 (cost - tokens) / rate는 토큰이 다시 찰 때까지의 시간이라 Retry-After에 그대로 씁니다. cost가 capacity보다 크면 영원히 통과하지 못하므로 미리 막습니다.
• threading.Lock은 읽기·계산·쓰기를 한 덩어리로 묶어 두 스레드가 같은 토큰을 쓰지 못하게 합니다.
• 창 카운터는 창 번호가 바뀔 때만 카운터를 옮기고, 두 칸 이상 비었으면 직전 값을 0으로 둡니다.

가짜 시계를 넣으면 테스트에서 결과가 늘 같게 나옵니다.

t = [0.0]
bucket = TokenBucket(rate=1, capacity=3, clock=lambda: t[0])
print([bucket.try_acquire()[0] for _ in range(4)])  # [True, True, True, False]
print(bucket.try_acquire())                          # (False, 1.0)
t[0] = 1.0
print(bucket.try_acquire())                          # (True, 0.0)

키마다 버킷이 필요하면 dict[str, TokenBucket]에 담아 둡니다. 가득 찬 버킷은 새로 만든 버킷과 구별되지 않으므로, 오래 쓰지 않은 키는 지워도 판정이 달라지지 않습니다.

C++

#include <algorithm>
#include <chrono>
#include <mutex>

class TokenBucket {
public:
    using Clock = std::chrono::steady_clock;

    TokenBucket(double rate, double capacity)
        : rate_(rate), capacity_(capacity), tokens_(capacity), updated_(Clock::now()) {}

    bool allow(double cost = 1.0) {
        std::lock_guard<std::mutex> lock(mutex_);
        auto now = Clock::now();
        std::chrono::duration<double> elapsed = now - updated_;
        tokens_ = std::min(capacity_, tokens_ + elapsed.count() * rate_);
        updated_ = now;
        if (tokens_ < cost) return false;
        tokens_ -= cost;
        return true;
    }

private:
    double rate_, capacity_, tokens_;
    Clock::time_point updated_;
    std::mutex mutex_;
};

// TokenBucket bucket(2.0, 5.0); allow()를 빠르게 8번 부르면 true 5번, false 3번

steady_clock은 Python의 monotonic처럼 뒤로 가지 않는 시계입니다.

Java

public final class TokenBucket {
    private final double rate, capacity;
    private double tokens;
    private long updatedNanos;

    public TokenBucket(double rate, double capacity) {
        this.rate = rate;
        this.capacity = capacity;
        this.tokens = capacity;
        this.updatedNanos = System.nanoTime();
    }

    public synchronized boolean allow(double cost) {
        long now = System.nanoTime();
        double elapsed = (now - updatedNanos) / 1_000_000_000.0;
        tokens = Math.min(capacity, tokens + elapsed * rate);
        updatedNanos = now;
        if (tokens < cost) return false;
        tokens -= cost;
        return true;
    }
}
// new TokenBucket(2.0, 5.0): allow(1.0)을 빠르게 8번 부르면 true 5번, false 3번

System.nanoTime()은 경과 시간 측정용 단조 시계이고, synchronized가 Python의 잠금 역할을 합니다.

TypeScript

export class TokenBucket {
  private tokens: number;
  private updated: number;

  constructor(
    private readonly rate: number,
    private readonly capacity: number,
    private readonly now: () => number = () => performance.now() / 1000,
  ) {
    this.tokens = capacity;
    this.updated = this.now();
  }

  allow(cost = 1): boolean {
    const now = this.now();
    this.tokens = Math.min(this.capacity, this.tokens + (now - this.updated) * this.rate);
    this.updated = now;
    if (this.tokens < cost) return false;
    this.tokens -= cost;
    return true;
  }
}

const bucket = new TokenBucket(2, 5);
console.log(Array.from({ length: 8 }, () => bucket.allow()));
// [true, true, true, true, true, false, false, false]

JavaScript는 한 스레드에서 이벤트 루프로 돌아가므로 잠금이 필요 없습니다. 여러 프로세스나 서버가 한도를 나눠 쓴다면 마지막 장의 Redis 방식으로 넘어가야 합니다.

정리

• 토큰 버킷은 토큰 수와 마지막 갱신 시각, 두 값만으로 구현됩니다.
• 슬라이딩 창 카운터는 창 번호와 카운터 두 개로 구현됩니다.
• 시계를 주입하면 시간에 의존하는 코드를 결정적으로 테스트할 수 있습니다.
• 여러 스레드가 같은 버킷을 쓰면 읽기·계산·쓰기를 잠금으로 묶어야 합니다.