有个做图片识别的朋友上个月找我帮忙看日志。他们一个 /v1/ocr 接口，同一天里同一个 IP 打了三十多万次，平均下来每秒四次，但峰值那几分钟能到两百多。服务本身没挂，问题是 GPU 队列被堵死了，正常用户的请求全在排队，客诉一堆。
他第一反应是上 Nginx 的 limit_req。我让他先别急，问了一句：你们的额度是按什么算的？他说免费用户每天 100 次，专业版每分钟 600 次，企业版不封顶但要求 QPS 不超过 50。这种按用户身份、按不同时间单位、不同口径的额度，Nginx 的 limit_req_zone 配不出来。所以这事只能落到应用层。
于是就有了下面这一摊。我把它完整写出来，包括中间写错的版本。
固定窗口的边界问题，我用一段代码复现了
最早他们自己写过一个最简单的：Redis 里存一个 INCR 计数器，key 是 rate:uid:当前分钟，超了 600 就拒绝。这就是标准的固定窗口。
我把这个逻辑抽出来单独跑了一下，问题立刻暴露。假设额度是每分钟 100 次，窗口在整分时切换。那么一个客户端在 12:00:59 打 100 次，窗口刷新，12:01:00 再打 100 次。两秒之内 200 个请求全部合法通过，是额度的两倍。
# 固定窗口的边界问题,用 asyncio 模拟一下
import asyncio, time
class FixedWindow:
def __init__(self, limit: int, window: int = 60):
self.limit = limit
self.window = window
self.count = 0
self.window_start = time.monotonic()
def allow(self) -> bool:
now = time.monotonic()
if now - self.window_start >= self.window:
self.count = 0
self.window_start = now
if self.count < self.limit:
self.count += 1
return True
return False
async def attack(fw: FixedWindow):
# 故意卡在窗口末尾打满,然后等窗口翻篇再打满
while time.monotonic() - fw.window_start < 59.5:
await asyncio.sleep(0.01)
passed = 0
for _ in range(100):
if fw.allow():
passed += 1
await asyncio.sleep(1.0) # 跨过窗口边界
for _ in range(100):
if fw.allow():
passed += 1
print("两秒内通过:", passed)
fw = FixedWindow(limit=100, window=60)
asyncio.run(attack(fw))跑出来是 200。这个数字看着不吓人，但如果这个接口是付费的、或者背后是 GPU，两秒放进来两倍的量就足够把队列打满。固定窗口的毛病不在实现，在它天生不具备「连续时间视野」。
为什么最后选了令牌桶
常见的三种算法我都试过：
- 滑动窗口日志。把每个请求的时间戳存进列表，判断最近 60 秒内的数量。精确，但内存占用跟请求量线性相关。上限 600 次的话，等于每个用户要存 600 个时间戳，一万个用户就是六百万个浮点数。不划算。
- 漏桶。请求先进队列，以固定速率流出。这个平滑是平滑，但它的语义是「排队」，不是「拒绝」。OCR 这个场景里，用户点了按钮三秒没反应就会重试，排队反而制造更多重复请求。
- 令牌桶。按速率往桶里补令牌，桶有容量上限，请求消耗令牌。它允许一定程度的突发（桶容量内），超出就立刻拒绝。
关键是最后那条：立刻拒绝。限流器的本职工作就是把超额的那部分挡在外面，而不是替调用方攒着。这点想清楚了，实现就简单很多。
第一版实现，被 time.time() 坑了
下面这版是我第一稿，逻辑上没问题，但有个隐蔽的坑：
import time
from dataclasses import dataclass, field
@dataclass
class TokenBucketV1:
rate: float # 每秒补充令牌数
capacity: int # 桶容量
_tokens: float = field(init=False)
_last: float = field(init=False)
def __post_init__(self):
self._tokens = float(self.capacity)
self._last = time.time() # 问题在这里
def allow(self, cost: int = 1) -> bool:
now = time.time()
elapsed = now - self._last
if elapsed > 0:
self._tokens = min(self.capacity, self._tokens + elapsed * self.rate)
self._last = now
if self._tokens >= cost:
self._tokens -= cost
return True
return False测试环境跑得好好的。上线第三天，运维做了一次 NTP 校时，机器时钟往回跳了 0.8 秒。这一跳导致 elapsed 变成负数，被我的 if elapsed > 0 挡了一下没出事——但如果时间往前跳呢？elapsed 变成 30 秒，tokens 直接被补满到 capacity，等于凭空发了一整桶令牌出去。
换个思路就更明显了：我真正关心的是「过了多久」，而不是「现在几点」。time.monotonic() 就是为这个存在的，它单调递增，不受系统时间调整影响。这一行改完，后面再没出现过莫名其妙的额度波动。
改完之后是这样：
import asyncio, time
from dataclasses import dataclass, field
@dataclass
class TokenBucket:
rate: float
capacity: int
_tokens: float = field(init=False)
_last: float = field(init=False)
def __post_init__(self):
self._tokens = float(self.capacity)
self._last = time.monotonic()
def _refill(self) -> None:
now = time.monotonic()
elapsed = now - self._last
if elapsed > 0:
self._tokens = min(self.capacity, self._tokens + elapsed * self.rate)
self._last = now
def try_acquire(self, cost: int = 1) -> bool:
self._refill()
if self._tokens >= cost:
self._tokens -= cost
return True
return False关于加锁：先搞清楚哪些代码是原子的
写完上面这版，我第一反应是「并发安全吗」，然后条件反射想在 try_acquire 外面包一层 asyncio.Lock。
后来意识到没必要。try_acquire 从头到尾没有一个 await，在 asyncio 的单线程事件循环里，它执行期间不可能被切走。整个函数就是一个原子操作。加锁不但没意义，反而引入了额外的调度开销。
但这个结论有前提。FastAPI 里有个很容易忽略的区别：用 async def 定义的端点跑在事件循环里，用普通 def 定义的端点会被自动丢进线程池执行。他们项目里有几个老接口是同步的，如果这些接口也共用同一个桶，那就真的是多线程访问同一份状态了。
我的处理是：限流这条路径只允许 async def，同步端点在入口处用 asyncio.to_thread 转一层，或者干脆给它单独配一个带锁的桶。把并发模型理清楚，比到处加锁省事。
别在桶里排队
网上能搜到的令牌桶实现大多长这样：拿不到令牌就 await asyncio.sleep()，睡一会儿再试。我一开始也这么写，压测的时候发现了问题。
假设桶空了，同时有 500 个协程在等。每个协程算出自己需要等待的时间，然后 sleep，同一时刻醒来，一起去抢那把锁（或者一起去改 _tokens）。醒来 500 个，只有 1 个能拿到令牌，剩下 499 个重新算时间再睡。这就是典型的惊群，CPU 大部分时间花在上下文切换上。
而且真实场景里这个等待毫无意义。用户点了个 OCR，你让他挂在那儿等 8 秒才给结果，体验还不如直接返回「稍后再试」。所以我把等待逻辑删了，只留 try_acquire，拿不到就是 429。
如果确实有内部任务需要排队（比如后台的批量转码），单独写一个 acquire，用 asyncio.Semaphore 控制在途数量，比在桶里自旋干净得多。
接进 FastAPI
先装依赖：
pip install fastapi uvicorn cachetools桶的存放需要一个容器。这里有个问题我一开始没注意：如果用普通 dict，key 是 plan:user_id，随着用户量增长它会无限膨胀。十万用户走一遍，就是十万个桶对象挂在内存里，而且大多数早就不会再用了。
用 cachetools.TTLCache 就行，带容量上限和过期时间，访问时自动续期：
from fastapi import FastAPI, Depends, HTTPException, Request
from cachetools import TTLCache
app = FastAPI()
# 最多缓存 10 万个桶,30 分钟没访问就回收
buckets: TTLCache = TTLCache(maxsize=100_000, ttl=1800)
# (每秒补充令牌数, 桶容量)
PLAN_LIMITS = {
"free": (100 / 86400, 20), # 每天 100 次,允许一次突发 20
"pro": (600 / 60, 60),
"enterprise": (50, 100),
}
def get_bucket(uid: str, plan: str) -> TokenBucket:
key = f"{plan}:{uid}"
bucket = buckets.get(key)
if bucket is None:
rate, cap = PLAN_LIMITS.get(plan, PLAN_LIMITS["free"])
bucket = TokenBucket(rate=rate, capacity=cap)
buckets[key] = bucket
return bucket
async def rate_limit(request: Request):
user = request.state.user # 由前面的认证中间件写入
bucket = get_bucket(user["id"], user["plan"])
if not bucket.try_acquire():
raise HTTPException(
status_code=429,
detail="请求过于频繁",
headers={"Retry-After": "1"},
)
@app.post("/v1/ocr", dependencies=[Depends(rate_limit)])
async def ocr(request: Request):
...
return {"ok": True}request.state.user 是我在认证中间件里挂上去的，做法很简单，@app.middleware("http") 里解完 JWT 之后 request.state.user = payload 就可以。这样限流器和业务代码都不需要重复解析 token。
还有一个细节：Retry-After 头我写死成 1 秒，其实可以算出来——桶空的时候，补一个令牌需要 1 / rate 秒。对 pro 用户来说是 0.1 秒，取整成 1 秒也够用。免费用户那个 86400 分之一秒的速率算出来是 864 秒，就得单独处理了，不然客户端会等一个半小时。这块我暂时没做，直接给的 1，但代码里留了注释，后面接 SDK 的时候要补。
多进程之后，额度变成了四倍
单进程跑着挺好，但他们是用 uvicorn --workers 4 起的。四个 worker 各自持有一份 buckets 字典，意味着免费用户每天 100 次的额度，实际能打 400 次。
这个问题的解法只有一个方向：把状态从进程内存挪到共享存储。Redis 是现成的。
直接把逻辑翻译成 Lua 脚本，用 Redis 的原子性保证正确性。但这里有个坑我踩了整整一个下午：Lua 里不能用 ARGV 传时间戳。
第一版我是这么写的：Python 端算好 time.monotonic() 传进去。问题是四台应用服务器启动时间不一样，monotonic 的基准点各不相同，传到 Lua 里彼此一比就乱套了。改用 time.time() 也不行，机器之间 NTP 同步总有几十毫秒误差，高并发下这几十毫秒足够多放出去一批令牌。
正确做法是让 Redis 自己给时间，用 TIME 命令，它在 Lua 里可以直接调用：
-- KEYS[1] = bucket key
-- ARGV[1] = rate(每秒令牌数) ARGV[2] = capacity
-- ARGV[3] = cost ARGV[4] = ttl 秒
local key = KEYS[1]
local rate = tonumber(ARGV[1])
local capacity = tonumber(ARGV[2])
local cost = tonumber(ARGV[3])
local ttl = tonumber(ARGV[4])
-- 用 Redis 服务器的时间,毫秒精度,所有节点看到的是同一个时钟
local t = redis.call('TIME')
local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)
local data = redis.call('HMGET', key, 'tokens', 'ts')
local tokens = tonumber(data[1])
local ts = tonumber(data[2])
if tokens == nil then
tokens = capacity
ts = now
end
local delta = now - ts
if delta < 0 then delta = 0 end
-- delta 是毫秒,rate 是每秒,所以要除以 1000
tokens = math.min(capacity, tokens + delta * rate / 1000)
local allowed = 0
if tokens >= cost then
tokens = tokens - cost
allowed = 1
end
redis.call('HSET', key, 'tokens', tostring(tokens), 'ts', tostring(now))
redis.call('EXPIRE', key, ttl)
return allowedPython 那边用 redis.asyncio 调，注意把脚本 register_script 缓存下来，不要每次都传全文上去：
import redis.asyncio as aioredis
class RedisTokenBucket:
def __init__(self, redis: aioredis.Redis, rate: float,
capacity: int, key_prefix: str = "rl:"):
self.redis = redis
self.rate = rate
self.capacity = capacity
self.prefix = key_prefix
self._sha = None
async def _load(self):
if self._sha is None:
self._sha = await self.redis.script_load(LUA_SRC)
return self._sha
async def try_acquire(self, uid: str, cost: int = 1) -> bool:
sha = await self._load()
try:
ok = await self.redis.evalsha(
sha, 1,
f"{self.prefix}{uid}",
self.rate, self.capacity, cost, 300,
)
except aioredis.ResponseError as e:
if "NOSCRIPT" not in str(e):
raise
# Redis 重启过,脚本缓存丢了,重新加载一次
self._sha = await self.redis.script_load(LUA_SRC)
ok = await self.redis.evalsha(
self._sha, 1,
f"{self.prefix}{uid}",
self.rate, self.capacity, cost, 300,
)
return bool(ok)那个 NOSCRIPT 分支不是防御性编程的洁癖。Redis 主从切换的时候，从库升主，脚本缓存是空的，第一批请求全会报这个错。我上线后第二周就遇上了。
压测的几个数字
下面这些是我在 4 核 8G 的云主机上跑的，uvicorn --workers 4，容量设 50，代价 1，用 ab -n 5000 -c 200 打，你机器不一样结果肯定不同，别当绝对值看。
| 实现方式 | 200 响应 | 429 响应 | 说明 |
|---|---|---|---|
| 单进程 + 内存桶 | 50 | 4950 | 数字精确，符合预期 |
| 4 进程 + 内存桶 | 201 | 4799 | 每进程一份桶，额度放大四倍 |
| 4 进程 + Redis Lua | 50 | 4950 | 共享状态，恢复精确 |
耗时上，单进程内存版 5000 个请求 QPS 在 9000 左右，Redis 版降到 2600 左右。这个差距主要是每请求一次网络往返带来的。如果对延迟敏感，可以在应用层加一层本地预检——先用内存桶挡掉明显超额的请求（比如本地桶已经空了就先拒），只有本地桶有余额时才去问 Redis。这样大部分被拒的请求根本不碰网络。代价是会有一定误拒，但限流器误拒几个请求，总比被刷爆强。
回过头看
这套东西写完，我最大的感受是：限流器的核心难点不在算法，在「想清楚你要什么语义」。是要排队、要平滑、还是要快速拒绝？这三种需求对应完全不同的实现，混在一起写出来的东西一定四不像。
另外就是那些细枝末节，time.monotonic、Redis 的 TIME、NOSCRIPT、TTLCache 的容量上限，每一个单拎出来都不算事，凑在一起就能让线上限流凭空多放或者少放一大截。这些代码现在都在跑，日志里 429 的曲线我看着还挺踏实。
如果你们也在做类似的东西，建议先别急着抄实现，把「额度到底按什么维度算」这件事在纸上推一遍。我见过太多项目限流器写得挺精致，结果 key 拼错了，用户 A 的额度扣在用户 B 头上。
