3. 令牌桶算法详解

令牌桶（Token Bucket）是业界应用最广泛的限流算法之一，尤其适合催收触达API这类需要容忍突发流量、同时保证长期平均速率的场景。本节从算法原理、参数配置、适用场景到Python伪代码实现，进行完整拆解。

3.1 算法原理

令牌桶的核心思想是：系统以恒定速率向一个桶中放入令牌，每个请求必须从桶中获取一个令牌才能被处理；如果桶中令牌不足，请求被拒绝或排队。桶的容量限制了最大突发流量。

3.2 参数配置与计算

令牌桶有两个核心参数，配置时需要结合催收触达API的SLA和上游系统容量进行推算。

关键计算公式：

• 最大突发请求数 = b（桶满时一次性放行）
• 突发后恢复时间 = (b - 当前令牌数) / r
• 实际吞吐量 ≤ min(请求到达率, r + 突发补偿)

3.3 适用场景分析

令牌桶算法在催收触达API中具有独特优势，尤其适合以下三类场景：

• 允许突发流量： 催收任务经常在整点批量触发（如上午10:00批量发送提醒），令牌桶允许短时超过平均速率，避免任务积压。
• 长期速率可控： 即使突发，长期平均速率仍被 r 严格限制，不会压垮下游短信/语音通道。
• 预热与冷启动： 桶初始为满，系统启动后即可处理突发请求，无需等待预热。

与其他限流算法的对比：

3.4 Python 伪代码实现

以下是一个生产级令牌桶的Python伪代码实现，包含时间戳驱动的令牌补充逻辑，避免使用定时器，性能更高。

import time

class TokenBucket:
    def __init__(self, rate: float, capacity: int):
        """
        :param rate: 令牌生成速率（个/秒）
        :param capacity: 桶容量（最大令牌数）
        """
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity          # 初始桶满，允许突发
        self.last_refill_time = time.time()

    def _refill(self):
        """根据时间差补充令牌"""
        now = time.time()
        elapsed = now - self.last_refill_time
        # 计算这段时间应生成的令牌数
        new_tokens = elapsed * self.rate
        # 更新令牌数，不超过容量
        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 表示限流拒绝
        """
        self._refill()
        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False

    def get_available_tokens(self) -> float:
        """查看当前可用令牌数"""
        self._refill()
        return self.tokens

# ========== 催收场景使用示例 ==========
# 初始化：速率 100 QPS，容量 200（允许瞬间 200 并发）
bucket = TokenBucket(rate=100, capacity=200)

# 模拟请求
for i in range(250):
    if bucket.try_acquire():
        print(f"请求 {i+1}: 通过 ✅")
    else:
        print(f"请求 {i+1}: 限流 ❌")
    time.sleep(0.005)  # 模拟 5ms 间隔