3. 令牌桶算法详解

令牌桶（Token Bucket）是限流与流量整形领域最经典的算法之一，在催收系统的外呼限流、AI 模型 Serving 的请求排队中均有广泛应用。其核心思想是：以恒定速率向桶中放入令牌，请求到达时必须获取到令牌才能被处理。桶的容量决定了最大突发流量，放入速率决定了长期平均处理速率。

3.1 算法原理

令牌桶由三个核心要素构成：

• 令牌生成器：以固定速率 r（单位：个/秒）向桶中添加令牌。
• 桶：最大容量 b（单位：个），用于存储未使用的令牌。
• 请求消费：每个请求到达时，从桶中取出 1 个（或 N 个）令牌；若桶中令牌不足，则请求被拒绝或等待。

当桶满时，新生成的令牌直接丢弃，不会累积。这种机制使得系统在空闲时能够“攒”下令牌，在突发流量到来时一次性释放，从而允许短时间内的流量尖峰，同时保证长期平均速率不超过 r。

3.2 参数配置：速率与容量

令牌桶的两个核心参数直接影响限流效果：

3.3 适用场景：突发流量

令牌桶最典型的应用场景是允许突发流量。以下对比表展示了不同场景下的适用性：

3.4 代码实现示例

以下是一个基于 Python 的令牌桶实现，支持线程安全操作，适用于催收外呼限流或 Model Serving 请求排队：

import time
import threading

class TokenBucket:
    """
    令牌桶限流器
    :param rate: 令牌放入速率（个/秒）
    :param capacity: 桶容量（最大令牌数）
    """
    def __init__(self, rate: float, capacity: int):
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity  # 初始时桶是满的，允许立即突发
        self.last_refill_time = time.monotonic()
        self.lock = threading.Lock()

    def _refill(self):
        """按时间间隔补充令牌"""
        now = time.monotonic()
        elapsed = now - self.last_refill_time
        new_tokens = elapsed * self.rate
        if new_tokens > 0:
            self.tokens = min(self.capacity, self.tokens + new_tokens)
            self.last_refill_time = now

    def try_acquire(self, tokens: int = 1) -> bool:
        """
        尝试获取 tokens 个令牌
        :return: True 表示获取成功，False 表示令牌不足
        """
        with self.lock:
            self._refill()
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

    def acquire(self, tokens: int = 1, timeout: float = None) -> bool:
        """
        阻塞获取令牌，直到获取成功或超时
        :param tokens: 需要的令牌数
        :param timeout: 超时时间（秒），None 表示一直等待
        :return: True 表示获取成功，False 表示超时
        """
        start_time = time.monotonic()
        while True:
            if self.try_acquire(tokens):
                return True
            if timeout is not None and (time.monotonic() - start_time) >= timeout:
                return False
            time.sleep(0.001)  # 短暂休眠，避免忙等

# 使用示例
if __name__ == "__main__":
    # 初始化：速率 10 个/秒，容量 20
    bucket = TokenBucket(rate=10, capacity=20)

    # 模拟 30 个并发请求
    for i in range(30):
        if bucket.try_acquire():
            print(f"请求 {i+1}: 通过")
        else:
            print(f"请求 {i+1}: 被限流")
        time.sleep(0.05)  # 模拟请求间隔 50ms

3.5 令牌桶 vs 漏桶：选型对比

在限流算法选型时，令牌桶与漏桶是最常被对比的两种方案。下表从多个维度进行对比：