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

本文深入剖析了Python中速率限制（Rate Limiting）的多维实现困境与最佳实践，明确指出仅依赖`functools.lru_cache`等内存级缓存机制在真实生产场景中完全不可行——它既无法支持时间窗口、用户/IP维度区分、请求阻塞和HTTP 429响应，更在多进程、分布式环境下彻底失效；文章对比了FastAPI中基于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 泛滥或者形同虚设。

今天带大家了解了的相关知识，希望对你有所帮助；关于文章的技术知识我们会一点点深入介绍，欢迎大家关注golang学习网公众号，一起学习编程~