Python中用Redis的ZSET实现滑动窗口限流,通过ZREMRANGEBYSCORE清理过期请求、ZADD插入新请求、ZCARD统计数量,并用Lua脚本保证原子性;支持按用户/IP/接口等多维度key设计,辅以降级与监控策略。

Python应用中用Redis实现滑动窗口限流,核心是利用Redis的ZSET(有序集合)按时间戳排序存储请求记录,并通过ZREMRANGEBYSCORE自动清理过期项,再用ZCARD统计当前窗口内请求数。不依赖外部库也能高效完成,关键是设计好时间粒度和key结构。
用ZSET模拟时间滑动窗口
ZSET的score设为毫秒级时间戳,member可设为唯一标识(如用户ID+时间戳或随机UUID),这样既能去重又能按时间排序。每次请求到来时:
- 先用
ZREMRANGEBYSCORE key -inf (current_timestamp - window_ms)剔除窗口外的旧记录 - 再用
ZADD key current_timestamp member插入新请求 - 最后用
ZCARD key获取当前窗口内请求数,与阈值比对
例如:1分钟内最多100次请求,window_ms = 60000,所有score小于time.time()*1000 - 60000的成员都会被自动清理。
避免单Key热点,支持多维度限流
直接用全局key会成为瓶颈,应按限流维度构造key,比如:
立即学习“Python免费学习笔记(深入)”;
- 按用户限流:
f"rate_limit:uid:{user_id}" - 按IP限流:
f"rate_limit:ip:{ip_address}" - 按接口+用户组合:
f"rate_limit:api:{endpoint}:uid:{user_id}"
注意key生命周期无需手动设置过期——只要窗口内无新请求,ZSET自然变空;也可给key加个较长的EXPIRE(如1小时),防止极端情况下残留空key堆积。
原子性保障:用Lua脚本封装操作
上面三步若拆成多个Redis命令,在高并发下可能因执行间隙导致超限(如A刚删完旧数据、B插入前A又插入一次)。推荐用Lua脚本一次性完成:
local key = KEYS[1] local now = tonumber(ARGV[1]) local window_ms = tonumber(ARGV[2]) local max_count = tonumber(ARGV[3])redis.call('zremrangebyscore', key, '-inf', '('..(now - window_ms)) local count = redis.call('zcard', key) if count < max_count then redis.call('zadd', key, now, ARGV[4]) return 1 else return 0 end
Python中调用:redis.eval(lua_script, 1, key, now_ms, window_ms, max_count, member_id),返回1表示放行,0表示拒绝。
补充技巧:降级与监控
生产环境别让限流失败直接抛异常:
- Redis异常时可降级为内存计数(如
threading.local或LRU cache),仅限临时兜底 - 对key做采样监控,比如每100次请求记录一次
ZCARD结果,观察分布是否倾斜 - 把“被限流次数”单独用
INCR计数,便于告警(如5分钟内超1000次触发通知)
基本上就这些。滑动窗口比固定窗口更平滑,比漏桶/令牌桶更易理解,用好ZSET和Lua,就能在Python服务里稳稳扛住突发流量。










