Python Web框架FastAPI怎么限制接口请求频率_结合SlowAPI与Redis实现IP防刷限流-Python教程-PHP中文网

直接用 slowapi 会失败，因其默认内存存储导致重启清零、多进程不共享；必须接入redis，且limiter不能在模块顶层初始化，需用depends异步注入。

python web框架fastapi怎么限制接口请求频率_结合slowapi与redis实现ip防刷限流

为什么直接用 `SlowAPI` 会失败？

很多人装完 slowapi 就写 Limiter，结果一跑就报 RedisConnectionError 或者限流完全不生效。根本原因是：SlowAPI 默认用内存存储计数器，重启服务后所有计数清零，且无法跨进程共享——生产环境多 worker（比如用 uvicorn --workers 4）时，每个进程各自计数，等于没限。

必须显式接入 Redis 才能真正生效。而接入前最容易忽略的一点是：SlowAPI 的 Limiter 实例**不能在模块顶层直接初始化**，否则 FastAPI 启动时还没连上 Redis 就会卡死或抛异常。

✅ 正确做法：用 Depends + 异步依赖注入，在每次请求时按需获取连接
❌ 错误写法：limiter = Limiter(key_func=get_remote_address, redis_url="redis://localhost") 放在文件顶部
注意 redis_url 必须带协议（redis:// 或 rediss://），不能只写 localhost:6379

怎么写一个可复用的限流依赖？

别把限流逻辑硬塞进每个路由函数里。应该封装成一个 Depends，支持按路径、IP、用户 ID 多维度控制。关键在于：用 key_func 动态生成限流 key，同时确保 Redis 连接由 FastAPI 的生命周期管理。

示例依赖定义：

from slowapi import Limiter
from slowapi.util import get_remote_address
from fastapi import Request, Depends
from redis.asyncio import Redis
<p>limiter = Limiter(key_func=get_remote_address)</p><div class="aritcle_card flexRow artxards">
											<div class="artcardd flexRow">
												<a class="aritcle_card_img" rel="nofollow" href="/xiazai/skill3894" title="Python数据分析(免费版)"><img
														src="https://img.php.cn/upload/skill/000/000/081/178982468260735.jpg" alt="Python数据分析(免费版)" onerror="this.onerror='';this.src='/static/lhimages/moren/morentu.png'" ></a>
												<div class="aritcle_card_info flexColumn">
													<a rel="nofollow" href="/xiazai/skill3894" title="Python数据分析(免费版)" class="overflowclass">Python数据分析(免费版)</a>
													<p class="overflowclass">提供Python数据清洗、统计分析与可视化建议，覆盖业务报表与科研数据的快速处理流程。</p>
												</div>
												<a rel="nofollow" href="/xiazai/skill3894" title="Python数据分析(免费版)" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span>
												</a>
											</div>
										</div><p>async def get_redis() -> Redis:
return Redis.from_url("redis://localhost", decode_responses=True)</p><p>def get_limiter():
return limiter
</p>

然后在路由中使用：

@app.get("/api/data")
@limiter.limit("5/minute")  # 每分钟最多 5 次
async def read_data(
    request: Request,
    redis: Redis = Depends(get_redis),
):
    return {"result": "ok"}

@limiter.limit("5/minute") 中的字符串格式必须是 N/second|minute|hour|day，不支持 5 per minute 这类自然语言
如果想按用户限流（如登录后 token 绑定），把 key_func 改成从 request.headers.get("Authorization") 提取 user_id
务必确认 Redis 服务已启动且端口可达，否则 get_redis() 会阻塞或超时

`429 Too Many Requests` 返回体怎么自定义？

默认返回的是纯文本 Too Many Requests，前端不好解析。要返回 JSON 并附带重试时间，得重写 http_429_handler。

FastAPI 启动前注册处理函数：

from slowapi.errors import RateLimitExceeded
from fastapi.responses import JSONResponse
<p>@app.exception_handler(RateLimitExceeded)
async def rate_limit_handler(request: Request, exc: RateLimitExceeded):
return JSONResponse(
status_code=429,
content={
"error": "rate_limited",
"retry_after": exc.retry_after,
"message": "请求过于频繁，请稍后再试"
}
)
</p>

exc.retry_after 是秒级整数，单位是秒（不是毫秒），可直接传给前端做倒计时
注意这个 handler 必须在 app = FastAPI() 之后、uvicorn.run() 之前注册
如果用了中间件修改响应头，要确保它不覆盖 X-RateLimit-Limit 和 X-RateLimit-Remaining 这些默认头

Redis 连接池和超时设置容易被忽略

线上高频接口下，没设连接池或超时会导致 Redis 连接耗尽、请求堆积甚至拖垮整个服务。SlowAPI 内部用的是 redis-py，但默认不启用连接池，也不设 socket 超时。

推荐初始化方式：Redis.from_url("redis://localhost", max_connections=20, socket_timeout=1, socket_connect_timeout=1)
max_connections 建议设为 uvicorn worker 数 × 4 左右（例如 4 个 worker → 设 16~20）
必须设 socket_timeout，否则 Redis 假死时请求会卡住几十秒，拖垮整个 FastAPI 实例
测试时可用 redis-cli monitor 观察实际发出的 INCR 和 EXPIRE 命令，确认 key 是否按预期生成

真正麻烦的从来不是写几行限流代码，而是 Redis 连接稳定性、key 设计粒度、以及多实例部署时的时钟漂移问题——比如不同服务器时间差 2 秒，可能导致限流窗口错位。这些细节不压测根本看不出来。