flask-limiter的token_bucket策略易失效,因默认MemoryStorage在多进程(如Gunicorn)下各worker独立维护令牌桶,导致限流完全失效;生产必须改用RedisStorage等共享存储,并规范rate格式与key_func。

为什么直接用 flask-limiter 的 token_bucket 策略容易失效?
因为默认的 token_bucket 实现依赖内存存储(MemoryStorage),在多进程或部署到 Gunicorn/UWSGI 时,每个 worker 维护独立桶,实际限流完全失效。你看到本地测试正常,上线后 QPS 翻倍甚至失控,大概率是这个原因。
实操建议:
立即学习“Python免费学习笔记(深入)”;
- 必须显式指定共享存储后端,如 Redis:
RedisStorage("redis://localhost:6379") - 避免用
@limiter.limit("100 per day")这类模糊表达,改用明确速率格式:"100/minute"或"5/second" - 若无 Redis,可用
MemcachedStorage,但别用MemoryStorage上生产
如何按用户 Token(而非 IP)做限流?
Flask-Limiter 默认使用 get_ipaddr 提取 key,要改成从请求头读取 Token 并哈希,否则所有用户共用一个桶。
实操建议:
立即学习“Python免费学习笔记(深入)”;
- 定义自定义 key_func:
def get_token_key(): auth = request.headers.get("Authorization") if auth and auth.startswith("Bearer "): token = auth[7:] return hashlib.sha256(token.encode()).hexdigest()[:16] return "anonymous" - 注册时传入:
@limiter.limit("10/minute", key_func=get_token_key) - 注意:不要直接返回原始 Token(泄露风险),也不要用完整哈希(key 过长影响 Redis 性能)
flask-limiter 的 key_func 在什么时机执行?
它在每次请求进入限流装饰器时调用,早于视图函数执行。这意味着你可以安全地访问 request、解析 header、查数据库(但不推荐——会拖慢限流路径)。
常见错误现象:
- 在
key_func里调用db.query.filter(...).first(),导致每个请求都查库,限流反而成性能瓶颈 - 误以为
key_func可以异步,实际 Flask-Limiter 不支持 async key_func(会报RuntimeWarning: coroutine was never awaited) - 返回
None或空字符串,Limiter 会 fallback 到 IP,行为不可控
如何让被限流的请求返回标准 JSON 错误?
默认返回 HTML 页面,API 场景下必须重写响应格式。
实操建议:
立即学习“Python免费学习笔记(深入)”;
- 注册错误处理器:
@limiter.request_filter def ip_whitelist(): return request.remote_addr in ["127.0.0.1"] @limiter.exempt def exempt_view(): pass @app.errorhandler(429) def ratelimit_handler(e): return {"error": "rate limit exceeded", "retry_after": e.description}, 429 -
e.description是 Limiter 内置的秒级等待时间(字符串),可直接透传 - 别漏掉
@limiter.request_filter—— 它比@app.before_request更早触发,适合白名单逻辑
@limiter.limit 注解全包圆。


















