3. 令牌桶算法详解

令牌桶算法（Token Bucket）是金融 API 限流中最常用的算法之一。它允许一定程度的突发流量，同时保证长期平均速率不超过设定阈值，非常适合处理交易、行情订阅等具有突发特征的金融请求。

3.1 工作机制

令牌桶的核心思想是：系统以恒定速率向一个“桶”中放入令牌，每个请求必须从桶中获取一个令牌才能被处理。如果桶中没有令牌，请求将被拒绝或等待。

3.2 参数配置

令牌桶算法有两个核心参数，直接决定限流效果：

3.3 Python 代码实现

以下是一个完整的令牌桶限流器实现，包含时间戳驱动的令牌补充逻辑：

import time
import threading

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()  # 上次补充时间
        self.lock = threading.Lock()    # 线程安全

    def _refill(self):
        """补充令牌"""
        now = time.time()
        elapsed = now - self.last_refill
        new_tokens = elapsed * self.rate
        self.tokens = min(self.capacity, self.tokens + new_tokens)
        self.last_refill = now

    def consume(self, tokens: int = 1) -> bool:
        """
        消费令牌
        :return: True 表示允许请求，False 表示限流
        """
        with self.lock:
            self._refill()
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

# 使用示例
if __name__ == "__main__":
    # 创建限流器：每秒 10 个令牌，容量 20
    bucket = TokenBucket(rate=10, capacity=20)

    # 模拟 30 个并发请求
    for i in range(30):
        allowed = bucket.consume()
        status = "✅ 通过" if allowed else "❌ 限流"
        print(f"请求 {i+1:2d}: {status} (剩余令牌: {bucket.tokens:.1f})")
        time.sleep(0.05)  # 模拟请求间隔

3.4 可视化分析

通过模拟请求并记录令牌变化，可以直观观察令牌桶的突发吸收能力：

3.5 金融场景适用性分析