17370845950

FastAPI 通过 slowapi 可实现基于内存的 IP 限流，无需 Redis；支持多粒度速率控制，但多进程下计数不共享、重启清零，适用于单进程轻量场景。

FastAPI 本身不内置限流功能，但可以借助 slowapi 这个轻量库实现基于 IP 的简单频率限制，且完全不依赖 Redis —— 它默认使用内存字典（dict）存储计数，适合低并发、开发或轻量生产场景。

1. 安装 slowapi 并初始化限流中间件

安装：

pip install slowapi

在 FastAPI 应用中启用限流中间件（注意：必须在创建 app 实例后尽早挂载）：

from fastapi import FastAPI
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.middleware import SlowAPIMiddleware

app = FastAPI()

# 使用内存存储（无 Redis）
limiter = Limiter(key_func=get_remote_address, default_limits=["5/minute"])
app.state.limiter = limiter
app.add_exception_handler(429, _rate_limit_exceeded_handler)
app.add_middleware(SlowAPIMiddleware)

2. 为路由添加 IP 级限流装饰器

直接在路由函数上使用 @limiter.limit，支持灵活的速率规则：

• "10/minute"：每分钟最多 10 次
• "100/hour"：每小时最多 100 次

  

• "5/5minutes"：每 5 分钟最多 5 次（需 slowapi ≥ 0.1.9）

示例：

@app.get("/public")
@limiter.limit("3/minute")
async def public_endpoint():
  return {"msg": "OK"}

3. 注意内存限流的关键限制与适用场景

该方案本质是进程内字典计数，因此：

• 多进程（如 gunicorn 多 worker）下各进程独立计数 → 实际限流阈值会放大（例如 4 个 worker，实际允许 4×设定值）
• 服务重启后计数清零 → 无持久化，不可用于强一致性要求的场景
• 适合单进程部署、内部工具接口、演示环境或流量极低的 API

若需跨进程/多实例统一限流，仍需 Redis 或其他共享存储 —— 此时 slowapi 也支持接入 Redis（通过 RedisDependency），但本方案明确规避了它。

4. 可选：自定义 key 函数以更精准识别客户端 IP

get_remote_address 默认取 request.client.host，但在反向代理（Nginx、Cloudflare）后可能拿到的是代理 IP。可改写 key 函数优先读取 X-Forwarded-For：

def get_client_ip(request):
  ip = request.headers.get("X-Forwarded-For", "").split(",")[0].strip()
  return ip or request.client.host

limiter = Limiter(key_func=get_client_ip, default_limits=["5/minute"])

⚠️ 生产中务必校验并信任可信代理头，避免伪造 IP 绕过限流。