Python速率限制中间件实现详解

本文深入剖析了Python中速率限制的常见误区与工程实践，指出仅依赖functools.lru_cache等内存级缓存机制无法满足真实场景需求——它既不支持时间窗口、用户/IP区分、请求阻塞，也无法保证多进程或分布式环境下的状态一致性；文章对比了FastAPI和Django生态中基于Redis的可靠限流方案（如fastapi-limiter、django-redis），强调滑动窗口实现细节、连接管理、原子计数、降级处理等关键要点，并警示纯内存字典实现因并发竞争、内存泄漏和清理失效而极易线上崩溃，最终点明：限流的核心挑战不在算法，而在跨系统边界下的状态一致性保障。

用 functools.lru_cache 做内存级速率限制够吗？

不够，除非你只跑单进程且请求量极低。它本质是函数调用缓存，不是限流——不区分用户、不计时间窗口、不阻塞超限请求，更不会返回 429 Too Many Requests。常见错误是把它当 rate_limit 用，结果压测时直接绕过限制。

• 真实限流必须带「时间窗口」和「计数器」逻辑，比如每分钟最多 100 次
• lru_cache 缓存的是返回值，不是「是否允许通过」的决策
• 多进程下各进程有独立缓存，完全失效；用 threading.Lock 补也解决不了跨进程问题
• 如果只是防自己写脚本误刷，那它凑合；但要防真实用户或爬虫，必须换方案

FastAPI 里怎么加 Redis 支持的限流中间件？

核心是拦截请求、查 Redis 计数、超限就中断。别自己手写连接池和 Lua 脚本，用 slowapi 或 fastapi-limiter 更稳。

• 推荐 fastapi-limiter：它封装了 aioredis（v2）或 redis-py（v4+），自动处理连接复用和原子计数
• 初始化时必须传 redis 实例，不是 URL 字符串：await limiter.init(redis)
• 限流装饰器要放在路由函数上，不是中间件里——中间件没提供 request.state 的干净入口点
• 注意 key_func 参数：默认按路径限流，想按用户 IP 限，得写 lambda request: request.client.host
• 错误现象：Redis 连接失败时默认静默放行，需手动检查 await limiter.get_connection() 并抛异常

Django 中间件里用 django-redis 实现滑动窗口限流要注意什么？

滑动窗口比固定窗口更准，但实现稍重。别用 zset 存所有时间戳——高频场景下会拖慢 Redis。

• 用 ZREMRANGEBYSCORE 清旧数据必须紧跟在 ZCARD 前，否则并发时可能漏删
• 单次请求的 key 要带毫秒级时间戳，比如 f"rate:{ip}:{int(time.time() * 1000)}"，不然窗口对不齐
• django-redis 的 get_client() 返回的是连接池实例，别每次 new 一个新连接
• 性能坑：如果用 cache.set(key, 1, timeout=60) 模拟窗口，实际是固定窗口，且无法做「当前窗口内已多少次」的精确查询
• 示例判断逻辑：if await redis.zcard(key) >= limit and await redis.zcount(key, now - window_ms, now) >= limit:

为什么用内存字典 + time.time() 实现的限流上线就崩？

因为没处理并发竞争和内存泄漏。看似简单，实则陷阱密集。

• 多线程下 dict 非线程安全，counter[uid] += 1 不是原子操作，会丢计数
• 时间窗口清理靠定时任务？Python 没可靠后台线程，Gunicorn 多 worker 下每个进程都得跑一份，极易错乱
• key 不设 TTL，字典无限增长，OOM 是早晚的事
• 正确做法：至少加 threading.RLock，且每个 key 必须配 last_clean_time 字段，每次访问前检查并清理过期项
• 更现实的选择：直接上 redis 或 memcached，本地内存只适合开发调试

限流的复杂点不在算法本身，而在「状态一致性」——跨进程、跨机器、网络分区、Redis 故障降级，这些边界情况一旦漏掉，线上就是 429 泛滥或者形同虚设。

到这里，我们也就讲完了《Python速率限制中间件实现详解》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注golang学习网公众号，带你了解更多关于的知识点！