[TOC] #### 1. 前言 --- 你的接口被恶意刷了，每秒几万次请求，服务器扛不住了 怎么办 ？限流 这一讲用 ZSet 实现一个滑动窗口限流器 #### 2. 实现 --- 每次请求往 ZSet 里加一个元素，score 是当前时间戳，member 用 uuid 保证唯一 然后统计窗口内有多少个元素，超过阈值就拒绝 ```python import uuid import time # 用 pipeline 批量执行，四条命令一次网络往返搞定 def is_allowed(key, limit=100, window=60): now = int(time.time()) pipe = redis.pipeline() pipe.zremrangebyscore(key, 0, now-window) # 清理窗口外的旧数据 pipe.zadd(key, {f"{uuid.uuid4()}": now}) # 添加当前请求 pipe.zcard(key) # 统计窗口内数量 pipe.expire(key, window) # 设过期时间，防止 key 无限增长 _, _, count, _ = pipe.execute() return count <= limit ``` 不同粒度 ```python # 按用户限流：每用户每分钟最多100次 is_allowed(f"rate:user:{user_id}", limit=100, window=60) # 按接口限流：每个接口每分钟最多1000次 is_allowed(f"rate:api:/order/create", limit=1000, window=60) # 按IP限流 is_allowed(f"rate:ip:{ip}", limit=200, window=60) ``` #### 3. 本文小结 --- 小总结： + ZSet + 时间戳 = 滑动窗口限流 + pipeline 批量执行，uuid 保证唯一 + zremrangebyscore 清理过期数据 + 可按用户、接口、IP 不同粒度限流