Python 令牌桶限流器

功能说明

TokenBucket 用单调时钟实现令牌桶：桶容量决定允许的突发量，refill_per_second 决定匀速补充速率。每次 allow 先按经过时间补充令牌，再尝试扣除成本；成功返回 (True, 0.0)，失败返回 (False, 等待秒数)，调用方可以据此做退避或直接拒绝。

关键代码

now = self.clock()
elapsed = max(0.0, now - self._updated)
self._tokens = min(self.capacity, self._tokens + elapsed * self.refill_per_second)
self._updated = now
...
missing = cost - self._tokens
return False, missing / self.refill_per_second

max(0.0, ...) 防止时钟回拨导致令牌被扣成负数。

使用注意

• 默认使用 time.monotonic，不受系统时间调整影响；示例通过可注入的 clock 便于测试。
• 锁只保护令牌计算，临界区很短；高并发时仍可能成为热点。
• 成本不能超过桶容量，否则请求永远无法通过，函数会直接报错。
• 只实现单进程限流；多实例部署需要 Redis 等共享存储。
• 返回的等待时间是最早可以成功的最短时间，不保证到点后一定成功。

验证方式

示例容量为 3、速率为每秒 2 个：前三次请求立即通过，第四次被拒绝并提示等待约 0.5 秒；时钟前进 0.5 秒后下一次请求通过。可以改变容量与速率，检查突发与匀速补充的边界。