令牌桶算法 Token Bucket：Python 异步高效限流实现

/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */
#content_views .toc,
/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */
#content_views.markdown_views > p:empty:has(+ .toc),
#content_views.markdown_views > .toc + p:empty,
/* 富文本旧版目录标记 */
#content_views.htmledit_views #main-toc,
#content_views.htmledit_views #hr-toc,
#content_views.htmledit_views p[id*=\”-toc\”] {
display: none !important;
}
/* 目录去掉后，紧跟的首个标题不再多出一块上边距 */
#content_views.markdown_views > .toc + h1,
#content_views.markdown_views > .toc + h2,
#content_views.markdown_views > .toc + h3,
#content_views.markdown_views > .toc + h4,
#content_views.markdown_views > .toc + p:empty + h1,
#content_views.markdown_views > .toc + p:empty + h2,
#content_views.markdown_views > .toc + p:empty + h3,
#content_views.markdown_views > .toc + p:empty + h4 {
margin-top: 0 !important;
}

令牌桶算法 Token Bucket：Python 异步高效限流实现

在微服务网关、大模型 API 转发层与 RAG 检索系统中，流量限流（Rate Limiting） 是防御下游服务过载、保障 SLA 稳定的第一道钢铁防线。

传统的限流算法中：

• 固定窗口计数器（Fixed Window）：在窗口临界切换的瞬间，会遭遇双倍流量冲击（临界突刺问题）；
• 漏桶算法（Leaky Bucket）：以恒定速率流出请求，虽然绝对平滑，但完全丧失了应对突发合规脉冲流量（Burst Traffic）的能力，导致突发正常请求被大量误杀。

令牌桶算法（Token Bucket） 凭借其“以恒定速率生成令牌、允许一定程度的突发消费、消费必须持有令牌”的优良特性，成为了高并发流控领域事实上的黄金标准。

如何在 Python 异步生态（asyncio）中，手写一个无线程锁、纳秒级时间戳平滑递推、支持动态突发容量与异步非阻塞等待的高性能令牌桶限流装饰器？

令牌桶算法的核心数学与时间递推模型

传统的令牌桶实现如果开一个后台定时器（loop.call_later）每秒往桶里塞令牌，在高并发成千上万个租户下，定时器队列会迅速拖垮事件循环。

生产级最优解是采用**“惰性时间递推计算（Lazy Timestamp Evaluation）”**：不需要任何常驻后台定时器，只在每次请求到达的瞬间，通过当前物理时间与上一次请求时间的差值，数学计算出这段时间内应该补齐多少个新令牌！

[ 请求到达时刻: Now ]
|
v
[ 计算时间差: Delta = Now – Last_Update_Time ]
|
v 补齐新令牌
[ 当前令牌数: Tokens = min( Capacity, Tokens + Delta * Refill_Rate ) ]
|
+—————————————+
| |
v (Tokens >= 1: 拥有充足令牌) v (Tokens < 1: 桶已空)
[ Tokens -= 1, 立即放行请求! ] [ 模式 A: 立即抛出 429 异常 (Drop) ]
[ 更新 Last_Update_Time = Now ] [ 模式 B: 精确计算需等待秒数并 await sleep (Smooth) ]

Python asyncio 纯异步高性能令牌桶完整实现

import asyncio
import time
from typing import Optional

class AsyncTokenBucket:
def __init__(self, rate: float, capacity: float):
"""
:param rate: 令牌填充速率 (每秒生成的令牌数，如 100.0 表示 100 QPS)
:param capacity: 令牌桶最大突发容量 (允许瞬间积攒的最大脉冲并发数，如 200.0)
"""
self.rate = float(rate)
self.capacity = float(capacity)
self.current_tokens = float(capacity)
self.last_update_time = time.monotonic()
# 使用 asyncio.Lock 保障在同一个事件循环内对单桶状态修改的协程安全
self._lock = asyncio.Lock()

async def acquire(self, tokens_needed: float = 1.0, wait: bool = True, timeout: Optional[float] = None) -> bool:
"""
获取指定数量的令牌
:param tokens_needed: 本次请求需要的令牌数 (支持大请求扣减更多令牌)
:param wait: 当令牌不足时，是阻塞等待还是立即返回失败
:param timeout: 最大允许等待超时时间（秒）
"""
async with self._lock:
now = time.monotonic()
elapsed = now – self.last_update_time
self.last_update_time = now

# 1. 惰性递推计算新生成的令牌
self.current_tokens = min(self.capacity, self.current_tokens + elapsed * self.rate)

# 2. 如果当前令牌充足，直接扣除并瞬间放行
if self.current_tokens >= tokens_needed:
self.current_tokens -= tokens_needed
return True

# 3. 如果不需要等待，直接返回失败 (用于非阻塞拒绝)
if not wait:
return False

# 4. 计算还差多少令牌，以及生成这些差额令牌所需的绝对时间（秒）
deficit = tokens_needed – self.current_tokens
wait_time = deficit / self.rate

# 5. 检查是否超出用户设定的最大容忍等待超时
if timeout is not None and wait_time > timeout:
return False

# 核心：预支令牌，并将令牌数扣为 0
self.current_tokens = 0.0

# 6. 在锁外进行异步非阻塞休眠，完全释放锁与其他协程的并发权！
await asyncio.sleep(wait_time)
return True

编写通用 FastAPI / ASGI 限流装饰器

from functools import wraps
from fastapi import HTTPException, Request

def rate_limit(rate: float = 50.0, burst: float = 100.0, max_wait_sec: float = 2.0):
# 为每个路由或每个租户绑定专属令牌桶
bucket = AsyncTokenBucket(rate=rate, capacity=burst)

def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
# 获取令牌，最多允许排队 2 秒
allowed = await bucket.acquire(tokens_needed=1.0, wait=True, timeout=max_wait_sec)
if not allowed:
# 超出突发承载且排队超时，返回标准的 429 状态码
raise HTTPException(
status_code=429,
detail="Too Many Requests: Rate limit exceeded, please retry later."
)
return await func(*args, **kwargs)
return wrapper
return decorator

# 业务路由使用示例
# @app.post("/v1/rag/query")
# @rate_limit(rate=100.0, burst=150.0, max_wait_sec=1.5)
# async def rag_query_endpoint(request: Request):
# return {"status": "ok"}

实测压测表现与对比

我们在 2000 QPS 突发脉冲流量下对比了固定窗口与令牌桶算法：

总结

令牌桶算法的精妙之处在于它对**突发性（Burst）与平稳性（Smoothing）**的完美兼顾。利用单调时间戳实现惰性递推计算，在锁外执行 asyncio.sleep 释放并发，是写出单机支撑数万 QPS 的高性能 Python 限流组件的标准工程典范。