4. 基于令牌桶的限流器实现：Python 实现 TokenBucket 类、参数配置与线程安全

令牌桶（Token Bucket）是金融 API 限流中最常用的算法之一。它允许一定程度的突发流量，同时保证长期平均速率不超过阈值，非常适合处理交易、行情订阅等具有突发特征的请求模式。本节将基于 Python 实现一个线程安全的 TokenBucket 类，并详细解析容量、速率等核心参数的设计。

4.1 令牌桶算法原理

令牌桶的核心思想是：系统以恒定速率向桶中放入令牌，每个请求需要消耗一个或多个令牌才能被处理。桶有最大容量限制，超出容量的令牌被丢弃。当桶中令牌不足时，请求被限流（等待或拒绝）。

4.2 TokenBucket 类设计与参数配置

在金融场景中，限流器需要精确控制速率和突发量。我们设计的 TokenBucket 类包含以下核心参数：

4.3 Python 实现：线程安全版本

金融 API 限流器通常运行在多线程环境中（如处理多个客户端请求），因此必须保证线程安全。我们使用 threading.Lock 来保护共享状态。

import threading
import time

class TokenBucket:
    """线程安全的令牌桶限流器"""

    def __init__(self, capacity: int, rate: float):
        """
        初始化令牌桶

        :param capacity: 桶的最大容量（最大突发请求数）
        :param rate: 令牌放入速率（个/秒）
        """
        if capacity <= 0 or rate <= 0:
            raise ValueError("capacity 和 rate 必须为正数")
        self.capacity = float(capacity)
        self.rate = rate
        self.tokens = float(capacity)  # 初始时桶是满的
        self.last_refill_time = time.monotonic()
        self._lock = threading.Lock()

    def _refill(self):
        """补充令牌（非线程安全，需在持有锁时调用）"""
        now = time.monotonic()
        elapsed = now - self.last_refill_time
        if elapsed > 0:
            new_tokens = elapsed * self.rate
            self.tokens = min(self.capacity, self.tokens + new_tokens)
            self.last_refill_time = now

    def consume(self, tokens: int = 1) -> bool:
        """
        尝试消耗指定数量的令牌

        :param tokens: 需要消耗的令牌数（每个请求消耗 1 个或多个）
        :return: True 表示允许请求，False 表示限流
        """
        if tokens <= 0:
            raise ValueError("消耗令牌数必须为正数")
        with self._lock:
            self._refill()
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

    def get_available_tokens(self) -> float:
        """获取当前可用令牌数（线程安全）"""
        with self._lock:
            self._refill()
            return self.tokens

4.4 线程安全设计详解

在多线程环境下，对 tokens 和 last_refill_time 的并发访问会导致竞态条件。我们的设计遵循以下原则：

• 细粒度锁：使用 threading.Lock 保护所有读写共享状态的操作。
• 延迟补充：不在独立线程中定时补充令牌，而是在每次 consume() 或 get_available_tokens() 调用时，根据时间差计算应补充的令牌数。这种方式避免了额外的线程开销，且精度更高。
• 单调时钟：使用 time.monotonic() 而非 time.time()，防止系统时间调整导致限流器行为异常。

4.5 参数配置策略与金融场景适配

不同的金融 API 接口对限流的要求不同。以下表格展示了常见场景的参数配置建议：

4.6 使用示例与验证

以下代码展示了如何实例化令牌桶并模拟请求限流：

# 实例化：容量 10，速率 2 个/秒
bucket = TokenBucket(capacity=10, rate=2.0)

# 模拟 15 个请求
for i in range(15):
    allowed = bucket.consume(1)
    status = "✅ 允许" if allowed else "❌ 拒绝"
    print(f"请求 {i+1}: {status}，剩余令牌: {bucket.get_available_tokens():.2f}")
    time.sleep(0.1)  # 模拟请求间隔 100ms

运行结果示例（前 10 个请求因桶满而全部通过，后续请求因令牌补充速度不足而被限流）：

请求 1: ✅ 允许，剩余令牌: 9.00
请求 2: ✅ 允许，剩余令牌: 8.00
...
请求 10: ✅ 允许，剩余令牌: 0.00
请求 11: ❌ 拒绝，剩余令牌: 0.20  # 仅补充了 0.2 个令牌
请求 12: ❌ 拒绝，剩余令牌: 0.40
请求 13: ❌ 拒绝，剩余令牌: 0.60
请求 14: ❌ 拒绝，剩余令牌: 0.80
请求 15: ✅ 允许，剩余令牌: 0.00  # 积累到 1 个令牌后允许