最简可行方案是固定窗口计数器:以用户ID等为key,用INCR+EXPIRE(原子执行)实现每分钟限流;存在临界问题时改用Lua+ZSET实现滑动窗口,需保证原子性、成员唯一性及定期清理空key。

Redis + Python 实现限流的最小可行方案
直接用 redis-py 配合固定窗口或滑动窗口逻辑就能跑起来,不需要引入 flask-limiter 或 slowapi 这类封装库——除非你已经卡在集成或分布式一致性上。
最常用、最不容易出错的是「固定窗口计数器」:每分钟只允许 100 次请求,就以当前分钟为 key,比如 "rate:uid_123:202405211430",每次请求 INCR,再 EXPIRE 60 秒。简单、快、Redis 原生命令全支持。
- key 设计必须包含能区分调用者的字段(如用户 ID、IP、API 路径),否则会全局共用配额
-
INCR和EXPIRE必须用 pipeline 或 Lua 脚本保证原子性,否则高并发下可能漏设过期时间 - 固定窗口有临界问题:比如第 59 秒来了 100 次,第 60 秒又来 100 次,实际 2 秒内放行了 200 次
用 Lua 脚本解决原子性与滑动窗口需求
Redis 单线程执行 Lua,天然避免竞态。想做「最近 60 秒最多 100 次」,就得用滑动窗口,典型做法是用 ZSET 存时间戳,每次请求前 ZREMRANGEBYSCORE 清旧数据,再 ZCARD 判数量,最后 ZADD 新记录。
但这个逻辑不能拆成多条命令发过去——中间可能被其他客户端插入。必须塞进一个 Lua 脚本里执行:
立即学习“Python免费学习笔记(深入)”;
local key = KEYS[1]
local now = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local max = tonumber(ARGV[3])
redis.call("ZREMRANGEBYSCORE", key, 0, now - window)
local count = redis.call("ZCARD", key)
if count < max then
redis.call("ZADD", key, now, now .. ":" .. math.random(1000, 9999))
redis.call("EXPIRE", key, window + 1)
return 1
else
return 0
end调用时传 KEYS=["rate:ip_192.168.1.1"],ARGV=[str(time.time()), "60", "100"]。注意 EXPIRE 时间要略大于窗口,防止 key 提前消失。
- 别用
time.time()的浮点秒,Lua 里tonumber可能截断;建议传int(time.time()) -
ZSET成员值必须唯一,否则重复ZADD不会新增,导致计数偏低;加随机后缀是最轻量解法 - 这个脚本没做
DEL清理空 key,长期运行会产生大量零成员ZSET,需定期扫描ZCARD == 0的 key 并删掉
Flask 中间件里怎么安全嵌入限流逻辑
不要在路由函数里手写 redis.eval() ——容易漏异常处理、难复用、混业务逻辑。应该抽成可插拔的装饰器或 before_request 钩子,且必须带 fallback 机制:Redis 挂了不能直接 500,得降级放行或走本地内存计数(如 lru_cache + threading.Lock)。
- 装饰器参数要明确区分「标识来源」:
key_func=lambda r: r.headers.get("X-Real-IP")比硬编码request.remote_addr更可控 - Redis 连接必须用连接池(
ConnectionPool),别每次 new 一个Redis()实例,否则 fd 耗尽 - 超时设置不能省:
socket_timeout=0.1,否则 Redis 延迟毛刺会让整个 API 请求卡住 - 本地降级建议用
functools.lru_cache(maxsize=1000)缓存 IP → 计数映射,配合time.time() // 60做简易窗口,不依赖外部服务
为什么不用 timeit 测限流性能,而要看 Redis INFO
单次 INCR 或 Lua 执行本身很快(INFO commandstats 里 cmdstat_eval 的 usec_per_call,以及 latency 监控。
- 如果
eval平均耗时突增到 5ms 以上,先查 Lua 脚本有没有redis.call("KEYS", ...)这种 O(N) 操作 - 高频限流 key 会集中在某个 Redis 分片上(尤其用 Redis Cluster 时),导致 skew,要用
HASH_TAG强制打散,比如"rate:{uid_123}:ip_192.168.1.1" - 本地开发用
redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru启动,避免测试时 key 爆满被踢导致限流失效
滑动窗口的精度和资源开销永远在博弈:ZSET 存 100 条记录和存 10000 条,内存差 10 倍,ZREMRANGEBYSCORE 耗时也不同。上线前得按真实 QPS 压测,而不是只看单次脚本耗时。










