用functools.wraps包裹、状态存Redis(或内存)、按IP+路径等请求标识计数,是通用限流装饰器最稳起点;因多进程下纯内存滑动窗口不同步,且时钟漂移与竞争会导致阈值失真。

直接说结论:用 functools.wraps 包裹、状态存 Redis(或内存)、按请求标识(如 IP + 路径)计数,是通用限流装饰器最稳的起点。
为什么不能只用 time.time() 做滑动窗口?
纯内存 + 时间戳实现滑动窗口,在多进程或多实例部署下完全失效——每个进程维护自己的时间桶,根本不同步。哪怕单进程,也扛不住高频并发下的时钟精度漂移和临界竞争。
- 典型错误现象:
RateLimitExceeded报错忽高忽低,压测时限流阈值像摆锤 - 真实场景:Flask/FastAPI 用
gunicorn --workers 4启动后,实际 QPS 是配置值的 4 倍 - 正确做法:把计数器和过期时间交给 Redis 的
INCR+EXPIRE原子组合,或用redis-py的pipeline封装
@rate_limit(limit=100, window=60) 这类装饰器怎么支持不同存储后端?
硬编码 Redis 连接会让单元测试难写、本地调试变麻烦。得把「存储行为」抽象成可插拔接口。
- 定义一个基类
RateLimiterBackend,含incr(key, expire)和get(key)方法 - 提供
MemoryBackend(用dict+threading.Lock,仅限单线程开发环境)和RedisBackend(生产首选) - 装饰器初始化时接收
backend参数,默认为RedisBackend.from_url("redis://localhost") - 注意:MemoryBackend 的
expire必须手动轮询清理,否则内存泄漏;RedisBackend 则依赖EXPIRE自动回收
如何提取可靠的请求标识(key)?
Key 设计错了,限流就形同虚设——要么误杀(同一用户被多个 key 限制),要么漏放(攻击者换头就绕过)。
- Web 框架里常见错误:只取
request.remote_addr→ Nginx 反代后全是127.0.0.1 - 正确链路:从
X-Forwarded-For取第一个非内网 IP,fallback 到request.remote_addr,再拼上request.path或request.endpoint - FastAPI 示例中需通过
Depends注入Request,Flask 则用request全局对象——装饰器得适配两种获取方式 - 敏感操作(如登录接口)建议加
user_id(若已认证),未认证则用user_agent + ip组合防脚本
装饰器里要不要抛异常?还是返回 HTTP 429?
抛异常看似干净,但 Web 框架对异常的处理路径不统一:Flask 默认 500,FastAPI 默认 500 或 422,都得额外写错误处理器。不如在装饰器内部直接控制响应。
- 推荐在装饰器里判断超限后,直接
return JSONResponse({"error": "rate limit exceeded"}, status_code=429)(FastAPI)或make_response(..., 429)(Flask) - 必须设置
Retry-After响应头,值为距离下次窗口开始的秒数,否则前端重试逻辑会瞎撞 - 别忘了在响应头加
X-RateLimit-Limit/X-RateLimit-Remaining,方便前端做 UI 降级(比如禁用按钮) - 注意:如果路由函数本身返回
StreamingResponse或文件下载,就不能中途 return 响应体,此时只能抛HTTPException(429)
真正难的不是写通逻辑,而是 Key 的语义一致性、后端切换时的过期策略对齐、以及跨服务调用时如何复用同一套限流上下文——这些细节不画图、不跑集成测试,光看代码根本发现不了。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!











