FastAPI接口限流实战：从原理到Redis实现

1. 为什么需要接口限流上周隔壁团队的服务又被刷爆了原因是某个新上线的接口没有做限流防护被脚本疯狂调用导致数据库连接池耗尽。这让我想起三年前自己踩过的类似坑——当时一个公开API被恶意爬虫以每秒200次的频率请求直接导致服务雪崩。从那时起我就养成了给所有对外接口加限流的习惯。FastAPI作为高性能框架虽然自带异步优势能扛住较高并发但任何服务都有其承载上限。合理的限流策略能防止单接口过载影响全局服务避免恶意用户耗尽服务器资源为不同业务等级的用户提供差异化QPS2. 主流限流方案对比2.1 令牌桶 vs 漏桶算法在项目里实测过两种经典算法令牌桶生产环境首选允许突发流量适合电商秒杀场景# 伪代码示例 if 当前令牌数 0: 消耗令牌() 处理请求() else: 返回429状态码漏桶更严格限流强制恒定速率适合金融支付场景2.2 存储层选型根据分布式需求有不同的实现方式方案优点缺点适用场景内存计数器零延迟重启丢失数据单机开发环境Redis Lua原子性保证需要维护Redis生产环境集群部署数据库记录可做精细化统计性能最差审计需求强的场景提示中小项目建议直接用Redis方案既保证性能又便于扩展3. 完整代码实现3.1 基于Redis的令牌桶先安装必要依赖pip install redis fastapi-rate-limiter核心限流中间件from fastapi import FastAPI, Request, HTTPException from fastapi.responses import JSONResponse import redis.asyncio as redis app FastAPI() redis_client redis.from_url(redis://localhost:6379) async def rate_limiter(request: Request, key: str user_ip, limit: int 10): # 获取客户端标识可按需改用用户ID等 identifier request.client.host # Redis Lua脚本保证原子性 script local key KEYS[1] local limit tonumber(ARGV[1]) local current tonumber(redis.call(GET, key) or 0) if current 1 limit then return 0 else redis.call(INCR, key) redis.call(EXPIRE, key, 60) return 1 end passed await redis_client.eval(script, 1, f{key}:{identifier}, limit) if not passed: raise HTTPException(status_code429, detailToo Many Requests) app.middleware(http) async def add_rate_limit(request: Request, call_next): try: await rate_limiter(request) response await call_next(request) except HTTPException as e: return JSONResponse( status_codee.status_code, content{detail: e.detail} ) return response3.2 多维度限流策略实际业务中往往需要更精细化的控制# 接口级限流 app.get(/api/v1/orders) rate_limit(times100, minutes5) async def get_orders(): pass # 用户级限流基于JWT app.get(/api/v1/profile) rate_limit(times30, minutes1, key_funclambda r: r.state.user_id) async def get_profile(): pass4. 生产环境注意事项4.1 性能优化要点使用连接池管理Redis连接Lua脚本要尽量精简上例仅9行对高频接口考虑本地缓存定期同步的二级限流4.2 常见踩坑记录时间窗口不同步曾经因为NTP服务器故障导致集群节点间时间不一致限流失效。解决方案# 改用Redis服务器时间 redis.call(TIME)[0] # 获取Redis当前时间戳热点Key问题当所有请求都命中同一个Redis Key时会导致性能下降。我们的优化方案# 对Key进行分片 shard_id hash(identifier) % 10 key frate_limit:{shard_id}:{identifier}突发流量处理某次促销活动导致正常用户也被限流后来改进为# 动态调整限流阈值 current_limit get_dynamic_limit_from_config_center()5. 高级技巧自适应限流在网关层集成机器学习模型根据历史流量自动调整阈值from prometheus_client import CollectorRegistry registry CollectorRegistry() REQUEST_DURATION Histogram( http_request_duration_seconds, HTTP request duration, [method, endpoint], registryregistry ) def adaptive_limit(): # 根据P99响应时间动态调整 p99 get_histogram_p99() if p99 1.0: # 响应时间超过1秒 return current_limit * 0.8 elif p99 0.3: # 系统空闲 return current_limit * 1.2这种方案在我们订单系统中成功将过载率降低了73%但实现成本较高建议百万级QPS以上的系统再考虑。