令牌桶是一种允许突发流量的限流算法，系统以固定速率向桶添加令牌，请求需获取令牌才能执行；python可用threading实现单机版，高并发分布式场景须用redis+lua保证原子性。
什么是令牌桶限流
令牌桶是一种经典的限流算法，核心思想是系统以固定速率向桶中添加令牌，每次请求需要先获取一个令牌才能执行。桶有最大容量，满了就不再添加；没令牌时请求被拒绝或排队等待。相比计数器（简单粗暴）和漏桶（恒定流出），令牌桶允许一定突发流量，更贴合真实业务场景。
Python中用threading实现简易令牌桶
适合单进程、非高并发场景（如内部工具接口、管理后台API）。关键点：用锁保护共享状态，避免多线程竞争。
-
维护两个变量：当前令牌数
self._tokens和上次填充时间self._last_fill - 每次请求前尝试补充令牌：根据间隔时间计算应新增数量，但不超过桶容量
- 获取令牌失败则返回False：可据此直接返回429 Too Many Requests
示例代码片段：
import time
import threading
<p>class TokenBucket:
def <strong>init</strong>(self, capacity: int, fill_rate: float):
self.capacity = capacity
self.fill_rate = fill_rate # tokens per second
self._tokens = capacity
self._last_fill = time.time()
self._lock = threading.Lock()</p><pre class='brush:python;toolbar:false;'>def _fill(self):
now = time.time()
elapsed = now - self._last_fill
new_tokens = elapsed * self.fill_rate
self._tokens = min(self.capacity, self._tokens + new_tokens)
self._last_fill = now
def consume(self, tokens: int = 1) -> bool:
with self._lock:
self._fill()
if self._tokens >= tokens:
self._tokens -= tokens
return True
return False结合Flask/FastAPI做接口限流
在Web框架中使用时，推荐封装为装饰器或中间件，避免每个路由重复写逻辑。
立即学习“Python免费学习笔记（深入）”；
-
Flask示例（装饰器）：创建
@rate_limit(limit=10, per=60)，内部调用上面的TokenBucket实例（注意：多worker需改用Redis后端） -
FastAPI示例（依赖注入）：定义
Depends(get_bucket)，把桶实例注入到路径操作函数中 - 关键提醒：单机多进程（如gunicorn多个worker）下，内存级桶不共享，必须用Redis等外部存储同步状态
生产环境建议用Redis+Lua实现原子性
高并发、分布式服务必须保证“检查+扣减”是原子操作，否则会出现超卖（超限）。Redis的单线程特性和Lua脚本支持完美解决这个问题。
- 用一个key存当前令牌数和最后更新时间（如哈希结构或拼接字符串）
- Lua脚本内完成：读取旧值 → 计算新令牌数 → 判断是否足够 → 扣减并更新 → 返回结果
- Python端只需调用
redis.eval(lua_script, ...)，一行拿到是否放行
这样既避免网络往返开销，又杜绝竞态条件，是线上首选方案。
