艾体宝干货 | 【Redis实用技巧#9】FastAPI + Redis + 滑动窗口：告别误伤，实现公平限流
是否想设计一套让用户感到公平的 API 限流规则？通过平滑流量，避免随机触发 429 错误，并借助 Redis 与真正的滑动窗口算法，实现足够健壮的限流执行，以适应复杂的生产环境。 如果限流器上线后立刻收到客诉，并非个例。事实上，大多数所谓“简单”的限流方案，其简单程度就如同将折叠椅当作简单梯子来用，平时凑合，但一旦出问题便可能是严重的故障，且往往发生在最不该出错的时刻。 正确的解决方式不是提高限流阈值，而是让限流规则更具公平性。 本文将演示如何为 FastAPI 与 Redis 搭建滑动窗口算法，避免边界峰值问题，减少误判，同时保持足以应对真实流量的性能。 为什么固定窗口会导致误判？ 滑动窗口是怎么工作的 有几种实现方式，但有一个非常优雅的 Redis 方案： 判断（每次请求时）： 核心架构：如何保证高并发下的准确性？ 这里的关键在于，“清理、计数、添加” 这一系列操作必须是原子的。否则，在超高并发下，多个请求可能同时通过检查，导致实际请求数超出限制。因此，我们选择使用 Redis Lua 脚本来保证原子性。 设计限流键：我们要限制“谁”？ 一个推荐的实践策略是： Redis Lua脚本（原子滑动窗口） 返回结果： 在 FastAPI 中的优雅集成 为什么这种方法能避免误判？ 上生产环境前，务必考虑的几点 制定故障降级策略 小结
最常见的“固定窗口”算法，比如“每分钟最多 60 次请求”，看似简单有效，却隐藏着一个致命缺陷：
假设一个用户在 12:00:59 这一刻瞬间发出了 60 次请求。
紧接着下一秒 12:01:00，计数器清零重置。
然后他又立刻发出 60 次请求。
结果就是：在短短 1 秒多的时间里，用户实际发出了 120 次请求，而你的限流器却认为完全合规。
更糟糕的是，固定窗口常常会惩罚那些在时间窗口边界附近正常操作的用户。比如用户在某一分钟的最后几秒和下一分钟的开头发送了两小批请求，就很容易被系统标记为“滥用”——即使他的行为完全没有恶意。
滑动窗口算法正是为了解决这个问题而生的。
它没有生硬的“时间桶”概念，也不会在整点时刻突然重置计数器。整个时间窗口是连续滑动的，就像一条移动的时间滑轨。[客户端请求] --> [FastAPI 应用 (依赖注入/中间件)]
|
|--- (原子化限流检查) ---|
V
[Redis 集群]
(Key: 用户标识:路由路径)
(Value: 有序集合 ZSET)
在 coding 前，先定义“公平”的含义。
这个脚本一次性完成了滑动窗口限流的所有逻辑：清理旧数据、判断是否超限、记录新请求。-- 参数说明：-- KEYS[1]: 限流键，例如 "rate_limit:user_123:/api/search"-- ARGV[1]: 当前时间戳（毫秒）-- ARGV[2]: 窗口大小（毫秒），如 60000-- ARGV[3]: 限制次数，如 60-- ARGV[4]: 键的过期时间（秒），应略大于窗口local current_time = tonumber(ARGV[1])local window_size = tonumber(ARGV[2])local max_requests = tonumber(ARGV[3])local key_ttl = tonumber(ARGV[4])-- 1. 移除窗口之外的所有旧时间戳
redis.call("ZREMRANGEBYSCORE", KEYS[1], 0, current_time - window_size)-- 2. 获取当前窗口内的请求数量local current_count = redis.call("ZCARD", KEYS[1])-- 3. 判断是否超限if current_count >= max_requests then-- 计算还需要多久才能重试（基于窗口内最早的请求）local oldest_request = redis.call("ZRANGE", KEYS[1], 0, 0, "WITHSCORES")local wait_time_ms = 0if oldest_request[2] then
wait_time_ms = (tonumber(oldest_request[2]) + window_size) - current_time
if wait_time_ms < 0 then wait_time_ms = 0 endend-- 返回：不允许，当前计数，需等待的毫秒数return {0, current_count, wait_time_ms}end-- 4. 未超限，记录本次请求
redis.call("ZADD", KEYS[1], current_time, tostring(current_time))-- 5. 刷新键的过期时间
redis.call("EXPIRE", KEYS[1], key_ttl)-- 返回：允许，新的计数，无需等待return {1, current_count + 1, 0}
此示例使用redis-py的异步客户端redis.asyncio，并将限流器作为依赖项应用。from fastapi import FastAPI, Request, HTTPException, Depends
import time
import redis.asyncio as redis
app = FastAPI(title="带滑动窗口限流的API服务")# 初始化异步Redis客户端
redis_client = redis.Redis(host="localhost", port=6379, decode_responses=False)# 将上面的Lua脚本内容粘贴在这里
LUA_SLIDING_WINDOW_SCRIPT = """
-- ... Lua脚本内容同上 ...
"""
_script_sha1 = None # 缓存脚本加载后返回的SHA1值# 限流配置
RATE_LIMIT_WINDOW = 60 # 时间窗口：60秒
RATE_LIMIT_MAX_REQS = 60 # 最大请求数：60次
KEY_EXPIRE_BUFFER = 120 # 键的过期时间（稍长于窗口，便于调试）def _get_current_ms():"""获取当前毫秒时间戳"""return int(time.time() * 1000)async def _ensure_script_loaded():"""确保Lua脚本已被加载到Redis服务器"""global _script_sha1
if _script_sha1 is None:
_script_sha1 = await redis_client.script_load(LUA_SLIDING_WINDOW_SCRIPT)async def sliding_window_rate_limiter(request: Request):"""
核心限流依赖项。
可被用于全局中间件或单个路由的 `dependencies=[Depends(sliding_window_rate_limiter)]`。
"""await _ensure_script_loaded()# 1. 构造限流对象的标识符# 优先使用API Key，否则使用客户端IP（根据你的认证体系调整）
api_key = request.headers.get("X-API-Key")
client_identifier = api_key if api_key else request.client.host
# 2. 可选：将请求路径也作为限流维度的一部分，实现更细粒度控制
request_path = request.url.path
redis_key = f"rate_limit:{client_identifier}:{request_path}"# 3. 原子化执行限流逻辑
result = await redis_client.evalsha(
_script_sha1,1, # 表示后面只有一个Key
redis_key,
_get_current_ms(),
RATE_LIMIT_WINDOW * 1000, # 转为毫秒
RATE_LIMIT_MAX_REQS,
KEY_EXPIRE_BUFFER
)
allowed, current_count, retry_after_ms = int(result[0]), int(result[1]), int(result[2])# 4. 如果被限流，抛出标准的429错误if not allowed:# 将毫秒转换为秒（向上取整，最少1秒）
retry_after_seconds = max(1, (retry_after_ms + 999) // 1000)raise HTTPException(
status_code=429,
detail={"code": "rate_limit_exceeded","message": "请求过于频繁，请稍后再试。","retry_after": retry_after_seconds,"limit": RATE_LIMIT_MAX_REQS,"window": RATE_LIMIT_WINDOW,},
headers={"Retry-After": str(retry_after_seconds),"X-RateLimit-Limit": str(RATE_LIMIT_MAX_REQS),"X-RateLimit-Remaining": "0","X-RateLimit-Reset": str(int(time.time()) + retry_after_seconds),})# 5. 请求通过，可以在此处将剩余次数等信息添加到响应头（可选）# response.headers["X-RateLimit-Remaining"] = str(RATE_LIMIT_MAX_REQS - current_count)return True# 在需要限流的路由上使用依赖项
@app.get("/api/v1/search", dependencies=[Depends(sliding_window_rate_limiter)])async def search_products(query: str):"""商品搜索接口，受滑动窗口限流保护。"""# 这里是你的业务逻辑...return {"results": [], "query": query}# 健康检查接口通常不需要限流
@app.get("/health")async def health_check():return {"status": "healthy"}
只要你部署了多个 FastAPI 实例，就必须使用 Redis 这类外部存储来做计数。各个Pod内存里的计数器互不干扰，限流就形同虚设。
除非是面向公众的、最基础的防护，否则尽量结合用户身份。一个公司的出口IP背后可能有成百上千的员工，一人犯错，全员被封，并不是一个合适的方式。
查询接口 和 数据导出接口 对服务器的压力差别很大。可以为不同接口设置不同的 (窗口, 次数) 组合，甚至引入更高级的 令牌桶算法 来应对复杂成本。
如果 Redis 挂了怎么办？
一个好的API限流器，不应该让守规矩的用户感到访问如同碰运气一般。通过 FastAPI + Redis + 滑动窗口 这个组合，可以获得的是一个行为可预测、边界处理平滑、反馈信息有用的限流方案。
![[WindHub] 一点福利 无限并发的一些模型1](https://xiaohack.oss-cn-zhangjiakou.aliyuncs.com/typecho/images/2026/01/01/20260101155148_695627946a477.png!mark)