前言在后端系统开发中接口限流是一个非常常见但容易被忽视的基础能力。很多接口在功能测试阶段看起来没有问题但一旦进入真实线上环境就可能遇到以下情况某个用户频繁刷新页面导致接口被大量请求某个爬虫短时间内高频访问接口某个客户端出现死循环不断重复调用服务秒杀、报名、查询等高并发场景下请求量瞬间暴涨第三方调用方没有做好重试控制导致接口压力被放大。如果系统没有限流机制很容易出现数据库压力过大、缓存被打穿、服务响应变慢甚至整个应用被拖垮。本文将使用FastAPI Redis实现一个轻量级接口限流方案重点讲清楚以下内容什么是接口限流常见限流算法有哪些如何用 Redis 实现固定窗口限流固定窗口算法有什么问题如何进一步实现滑动窗口限流如何在 FastAPI 中封装成通用依赖真实项目中需要注意哪些细节。本文不依赖复杂框架重点放在底层逻辑和工程实现。一、什么是接口限流接口限流的核心目标是限制某个用户、某个 IP、某个接口或某类资源在单位时间内的访问次数。例如同一个 IP 每分钟最多访问登录接口 10 次 同一个用户每秒最多提交订单 3 次 同一个接口每分钟最多允许 1000 次请求当请求超过限制后系统可以直接返回错误例如{ code: 429, message: Too Many Requests }HTTP 协议中429 Too Many Requests就是专门用于表示请求过于频繁的状态码。二、常见限流算法常见接口限流算法主要有以下几种。1. 固定窗口算法固定窗口算法是最简单的限流方式。例如限制每个 IP 每 60 秒最多请求 100 次系统会以 60 秒为一个窗口记录当前窗口内的请求次数。如果请求次数小于 100则允许访问。如果请求次数超过 100则拒绝访问。优点是实现简单性能好。缺点是存在窗口边界问题。例如12:00:59 请求 100 次 12:01:00 请求 100 次从每分钟限制来看这两批请求分别落在两个窗口内都没有超限。但从真实时间来看用户在 1 秒左右发送了 200 次请求。这就会造成瞬时流量突刺。2. 滑动窗口算法滑动窗口算法会记录一段时间内的真实请求时间戳。例如限制任意 60 秒内最多请求 100 次每次请求进来时系统会先删除 60 秒之前的请求记录然后统计最近 60 秒内的请求数量。如果数量小于 100则允许访问并记录当前请求时间。如果数量已经达到 100则拒绝访问。滑动窗口比固定窗口更精确但实现成本和 Redis 操作成本也更高。3. 令牌桶算法令牌桶算法会以固定速率向桶中放入令牌。每次请求都需要消耗一个令牌。如果桶中有令牌则允许请求。如果桶中没有令牌则拒绝请求或等待。令牌桶适合允许一定程度的突发流量。4. 漏桶算法漏桶算法把请求看作进入桶中的水请求会以固定速率流出。如果请求进入速度太快桶满后新请求就会被丢弃。漏桶更强调平滑流量适合控制下游系统的处理速率。三、环境准备本文使用以下技术栈Python 3.10 FastAPI Redis redis-py uvicorn安装依赖pip install fastapi uvicorn redis如果本地没有 Redis可以使用 Docker 快速启动docker run -d \ --name redis-limiter \ -p 6379:6379 \ redis:7创建项目目录rate_limit_demo/ ├── main.py ├── redis_client.py ├── limiter.py └── requirements.txt四、连接 Redis先创建redis_client.pyimport redis redis_client redis.Redis( hostlocalhost, port6379, db0, decode_responsesTrue )这里使用同步 Redis 客户端方便演示。在高并发异步项目中也可以使用redis.asyncio。五、实现固定窗口限流固定窗口限流的核心思路是每个限流对象生成一个 Redis key 每次请求让 key 对应的计数器加 1 如果是第一次访问设置过期时间 如果计数超过阈值则拒绝请求例如 key 可以设计为rate_limit:login:127.0.0.1表示某个 IP 对登录接口的访问次数。创建limiter.pyimport time from fastapi import HTTPException, Request from redis_client import redis_client def fixed_window_limiter( key: str, limit: int, window_seconds: int ): 固定窗口限流 key: 限流 key limit: 窗口内最大请求次数 window_seconds: 窗口长度单位秒 current redis_client.incr(key) if current 1: redis_client.expire(key, window_seconds) if current limit: raise HTTPException( status_code429, detailToo Many Requests ) return True这个实现非常简单但有一个细节需要注意current redis_client.incr(key)INCR是 Redis 原子操作因此并发场景下不会出现多个请求同时读取相同旧值的问题。六、在 FastAPI 接口中使用固定窗口限流创建main.pyfrom fastapi import FastAPI, Request from limiter import fixed_window_limiter app FastAPI() app.get(/api/search) def search(request: Request): client_ip request.client.host key frate_limit:search:{client_ip} fixed_window_limiter( keykey, limit5, window_seconds60 ) return { message: search success }启动项目uvicorn main:app --reload访问接口curl http://127.0.0.1:8000/api/search如果同一个 IP 在 60 秒内访问超过 5 次就会返回{ detail: Too Many Requests }状态码为429。七、固定窗口算法的问题固定窗口虽然实现简单但存在明显的边界问题。假设限制规则是每分钟最多请求 5 次用户在第 59 秒请求 5 次然后在第 60 秒之后马上又请求 5 次。从 Redis 计数窗口来看这 10 次请求分别属于两个周期因此都可能被允许。但从实际情况看用户在极短时间内完成了 10 次请求仍然可能对服务造成压力。这就是固定窗口算法的典型缺陷。为了解决这个问题可以使用滑动窗口算法。八、用 Redis ZSET 实现滑动窗口限流滑动窗口需要记录每次请求的时间戳。Redis 的有序集合 ZSET 很适合这个场景。可以把请求时间戳作为 score把请求唯一标识作为 member。每次请求时执行以下步骤1. 删除窗口之前的请求记录 2. 统计当前窗口内的请求数量 3. 如果数量达到限制则拒绝请求 4. 如果没有达到限制则写入当前请求记录 5. 设置 key 过期时间例如限制任意 60 秒内最多请求 5 次当前时间是now则需要保留now - 60 秒 到 now这段时间内的请求记录。代码如下import time import uuid from fastapi import HTTPException from redis_client import redis_client def sliding_window_limiter( key: str, limit: int, window_seconds: int ): 滑动窗口限流 key: 限流 key limit: 窗口内最大请求次数 window_seconds: 窗口长度单位秒 now time.time() window_start now - window_seconds redis_client.zremrangebyscore(key, 0, window_start) current_count redis_client.zcard(key) if current_count limit: raise HTTPException( status_code429, detailToo Many Requests ) member f{now}:{uuid.uuid4()} redis_client.zadd(key, {member: now}) redis_client.expire(key, window_seconds) return True这里使用了几个 Redis 命令ZREMRANGEBYSCORE删除指定分数范围内的元素 ZCARD统计集合中元素数量 ZADD添加元素 EXPIRE设置过期时间每一次请求都会清理过期数据然后检查最近窗口内的请求数量。九、滑动窗口的并发问题上面的滑动窗口实现逻辑是正确的但在极高并发场景下仍然存在一个问题删除旧数据 统计数量 写入新请求这三个步骤不是一个原子操作。如果多个请求同时进来可能都在ZCARD时看到未超限然后一起写入导致短时间内超出限制。为了让操作原子化可以使用 Redis Lua 脚本。十、使用 Lua 脚本保证原子性Redis 执行 Lua 脚本时会把脚本作为一个整体执行。在脚本执行期间不会被其他命令插入。因此适合处理限流这种多步骤操作。Lua 脚本如下SLIDING_WINDOW_LUA local key KEYS[1] local now tonumber(ARGV[1]) local window tonumber(ARGV[2]) local limit tonumber(ARGV[3]) local member ARGV[4] local window_start now - window redis.call(ZREMRANGEBYSCORE, key, 0, window_start) local current redis.call(ZCARD, key) if current limit then return 0 end redis.call(ZADD, key, now, member) redis.call(EXPIRE, key, window) return 1 封装成 Python 函数import time import uuid from fastapi import HTTPException from redis_client import redis_client SLIDING_WINDOW_LUA local key KEYS[1] local now tonumber(ARGV[1]) local window tonumber(ARGV[2]) local limit tonumber(ARGV[3]) local member ARGV[4] local window_start now - window redis.call(ZREMRANGEBYSCORE, key, 0, window_start) local current redis.call(ZCARD, key) if current limit then return 0 end redis.call(ZADD, key, now, member) redis.call(EXPIRE, key, window) return 1 def sliding_window_limiter_lua( key: str, limit: int, window_seconds: int ): now time.time() member f{now}:{uuid.uuid4()} allowed redis_client.eval( SLIDING_WINDOW_LUA, 1, key, now, window_seconds, limit, member ) if allowed ! 1: raise HTTPException( status_code429, detailToo Many Requests ) return True这样清理、统计、判断、写入就变成了一个原子操作。十一、封装成 FastAPI 依赖在实际项目中不建议每个接口都手动写限流逻辑。可以将限流器封装成 FastAPI 依赖。from fastapi import Request def rate_limit( limit: int, window_seconds: int, prefix: str ): def dependency(request: Request): client_ip request.client.host path request.url.path key frate_limit:{prefix}:{path}:{client_ip} sliding_window_limiter_lua( keykey, limitlimit, window_secondswindow_seconds ) return True return dependency使用方式如下from fastapi import FastAPI, Depends app FastAPI() app.get(/api/search, dependencies[Depends(rate_limit( limit5, window_seconds60, prefixsearch ))]) def search(): return { message: search success } app.post(/api/login, dependencies[Depends(rate_limit( limit3, window_seconds60, prefixlogin ))]) def login(): return { message: login success }这样不同接口就可以设置不同的限流规则。例如搜索接口每分钟 5 次 登录接口每分钟 3 次 提交接口每分钟 10 次十二、按用户 ID 限流上面的代码是按 IP 限流。但在登录后的业务系统中更常见的是按用户 ID 限流。例如同一个用户每分钟最多提交 10 次订单可以从请求头或认证信息中获取用户 ID。示例def user_rate_limit( limit: int, window_seconds: int, prefix: str ): def dependency(request: Request): user_id request.headers.get(X-User-Id) if not user_id: raise HTTPException( status_code401, detailUnauthorized ) path request.url.path key frate_limit:{prefix}:{path}:user:{user_id} sliding_window_limiter_lua( keykey, limitlimit, window_secondswindow_seconds ) return True return dependency接口中使用app.post(/api/order, dependencies[Depends(user_rate_limit( limit10, window_seconds60, prefixorder ))]) def create_order(): return { message: order created }需要注意的是真实项目中用户 ID 不应该直接相信请求头而应该从认证系统、JWT、Session 或网关传递的可信身份中获取。十三、返回更友好的错误信息默认返回{ detail: Too Many Requests }对于前端来说可能不够友好。可以统一返回结构from fastapi import HTTPException def raise_rate_limit_error(): raise HTTPException( status_code429, detail{ code: RATE_LIMITED, message: 请求过于频繁请稍后再试 } )在限流失败时调用if allowed ! 1: raise_rate_limit_error()这样前端可以根据code做统一处理。十四、增加 Retry-After 响应头HTTP 中可以使用Retry-After告诉客户端多久后可以重试。示例from fastapi import HTTPException def raise_rate_limit_error(retry_after: int): raise HTTPException( status_code429, detail{ code: RATE_LIMITED, message: 请求过于频繁请稍后再试 }, headers{ Retry-After: str(retry_after) } )例如Retry-After: 60表示客户端可以在 60 秒后重试。在更精细的滑动窗口实现中可以根据最早请求时间计算剩余等待时间。十五、限流 Key 的设计限流效果很大程度上取决于 key 的设计。常见 key 维度包括按 IP 限流 rate_limit:ip:{ip} 按用户限流 rate_limit:user:{user_id} 按接口限流 rate_limit:path:{path} 按用户 接口限流 rate_limit:user:{user_id}:path:{path} 按 IP 接口限流 rate_limit:ip:{ip}:path:{path}不同场景适合不同粒度。登录接口适合按 IP 账号共同限流rate_limit:login:ip:{ip} rate_limit:login:account:{account}这样可以防止同一个 IP 爆破多个账号也可以防止多个 IP 爆破同一个账号。下单接口适合按用户 ID 限流rate_limit:order:user:{user_id}公开查询接口适合按 IP 限流rate_limit:query:ip:{ip}第三方开放接口适合按 API Key 限流rate_limit:openapi:apikey:{api_key}十六、真实项目中的注意事项1. 不要只在应用层限流应用层限流很灵活但在高流量场景下请求已经打到了应用服务。如果流量特别大应该在更靠前的位置限流例如网关层 负载均衡层 Nginx API Gateway 服务网格应用层限流更适合做精细化业务控制。2. 注意代理场景下的真实 IP如果服务部署在反向代理后面直接使用request.client.host拿到的可能是代理服务器 IP而不是真实客户端 IP。这种情况下需要读取X-Forwarded-For X-Real-IP但这些请求头也可能被伪造因此必须由可信网关统一注入和清洗。3. Redis 故障时如何处理限流依赖 Redis。如果 Redis 不可用系统需要决定采用哪种策略fail-openRedis 故障时放行请求 fail-closedRedis 故障时拒绝请求大多数普通业务接口更适合 fail-open避免因为 Redis 故障导致主业务全部不可用。但对于登录、支付、风控等敏感接口可以根据业务风险选择更严格策略。4. 避免 Redis key 爆炸如果 key 维度过细例如用户 ID IP 接口 设备 参数可能会产生大量 Redis key。因此 key 设计需要平衡精度和资源消耗。建议设置合理过期时间避免把长参数直接拼进 key必要时对参数做 hash定期监控 Redis key 数量和内存占用。5. 限流不是权限控制限流只能限制访问频率不能替代认证和权限校验。例如限流解决的是“请求太多”的问题 认证解决的是“你是谁”的问题 权限解决的是“你能不能访问”的问题三者不能混为一谈。十七、完整示例代码下面给出一个简化版完整示例。import time import uuid import redis from fastapi import FastAPI, Request, Depends, HTTPException redis_client redis.Redis( hostlocalhost, port6379, db0, decode_responsesTrue ) SLIDING_WINDOW_LUA local key KEYS[1] local now tonumber(ARGV[1]) local window tonumber(ARGV[2]) local limit tonumber(ARGV[3]) local member ARGV[4] local window_start now - window redis.call(ZREMRANGEBYSCORE, key, 0, window_start) local current redis.call(ZCARD, key) if current limit then return 0 end redis.call(ZADD, key, now, member) redis.call(EXPIRE, key, window) return 1 def raise_rate_limit_error(): raise HTTPException( status_code429, detail{ code: RATE_LIMITED, message: 请求过于频繁请稍后再试 } ) def sliding_window_limiter_lua( key: str, limit: int, window_seconds: int ): now time.time() member f{now}:{uuid.uuid4()} allowed redis_client.eval( SLIDING_WINDOW_LUA, 1, key, now, window_seconds, limit, member ) if allowed ! 1: raise_rate_limit_error() return True def rate_limit( limit: int, window_seconds: int, prefix: str ): def dependency(request: Request): client_ip request.client.host path request.url.path key frate_limit:{prefix}:{path}:{client_ip} sliding_window_limiter_lua( keykey, limitlimit, window_secondswindow_seconds ) return True return dependency app FastAPI() app.get(/api/search, dependencies[Depends(rate_limit( limit5, window_seconds60, prefixsearch ))]) def search(): return { message: search success } app.post(/api/submit, dependencies[Depends(rate_limit( limit3, window_seconds60, prefixsubmit ))]) def submit(): return { message: submit success }启动服务uvicorn main:app --reload测试接口curl http://127.0.0.1:8000/api/search连续请求超过限制后会返回 429。十八、总结本文从零实现了一个基于 FastAPI 和 Redis 的接口限流方案。核心流程可以概括为请求进入接口 ↓ 根据 IP、用户 ID 或 API Key 生成限流 key ↓ 在 Redis 中记录请求时间或请求次数 ↓ 判断是否超过限制 ↓ 允许请求或返回 429固定窗口算法实现简单适合基础场景但存在窗口边界问题。滑动窗口算法更加精确可以限制“任意时间段内”的请求次数更适合对流量控制要求较高的接口。在工程实践中限流方案需要结合业务场景选择合适粒度公开接口可以按 IP 限流用户接口可以按用户 ID 限流登录接口可以按 IP 和账号共同限流开放平台接口可以按 API Key 限流高并发核心接口可以放在网关层提前限流。接口限流不是一个复杂功能但它是后端系统稳定性建设中非常重要的一环。一个好的限流设计既能保护系统不被异常流量拖垮也能让正常用户在高峰期获得更稳定的访问体验。
FastAPI + Redis 实现接口限流：从固定窗口到滑动窗口的完整实践
前言在后端系统开发中接口限流是一个非常常见但容易被忽视的基础能力。很多接口在功能测试阶段看起来没有问题但一旦进入真实线上环境就可能遇到以下情况某个用户频繁刷新页面导致接口被大量请求某个爬虫短时间内高频访问接口某个客户端出现死循环不断重复调用服务秒杀、报名、查询等高并发场景下请求量瞬间暴涨第三方调用方没有做好重试控制导致接口压力被放大。如果系统没有限流机制很容易出现数据库压力过大、缓存被打穿、服务响应变慢甚至整个应用被拖垮。本文将使用FastAPI Redis实现一个轻量级接口限流方案重点讲清楚以下内容什么是接口限流常见限流算法有哪些如何用 Redis 实现固定窗口限流固定窗口算法有什么问题如何进一步实现滑动窗口限流如何在 FastAPI 中封装成通用依赖真实项目中需要注意哪些细节。本文不依赖复杂框架重点放在底层逻辑和工程实现。一、什么是接口限流接口限流的核心目标是限制某个用户、某个 IP、某个接口或某类资源在单位时间内的访问次数。例如同一个 IP 每分钟最多访问登录接口 10 次 同一个用户每秒最多提交订单 3 次 同一个接口每分钟最多允许 1000 次请求当请求超过限制后系统可以直接返回错误例如{ code: 429, message: Too Many Requests }HTTP 协议中429 Too Many Requests就是专门用于表示请求过于频繁的状态码。二、常见限流算法常见接口限流算法主要有以下几种。1. 固定窗口算法固定窗口算法是最简单的限流方式。例如限制每个 IP 每 60 秒最多请求 100 次系统会以 60 秒为一个窗口记录当前窗口内的请求次数。如果请求次数小于 100则允许访问。如果请求次数超过 100则拒绝访问。优点是实现简单性能好。缺点是存在窗口边界问题。例如12:00:59 请求 100 次 12:01:00 请求 100 次从每分钟限制来看这两批请求分别落在两个窗口内都没有超限。但从真实时间来看用户在 1 秒左右发送了 200 次请求。这就会造成瞬时流量突刺。2. 滑动窗口算法滑动窗口算法会记录一段时间内的真实请求时间戳。例如限制任意 60 秒内最多请求 100 次每次请求进来时系统会先删除 60 秒之前的请求记录然后统计最近 60 秒内的请求数量。如果数量小于 100则允许访问并记录当前请求时间。如果数量已经达到 100则拒绝访问。滑动窗口比固定窗口更精确但实现成本和 Redis 操作成本也更高。3. 令牌桶算法令牌桶算法会以固定速率向桶中放入令牌。每次请求都需要消耗一个令牌。如果桶中有令牌则允许请求。如果桶中没有令牌则拒绝请求或等待。令牌桶适合允许一定程度的突发流量。4. 漏桶算法漏桶算法把请求看作进入桶中的水请求会以固定速率流出。如果请求进入速度太快桶满后新请求就会被丢弃。漏桶更强调平滑流量适合控制下游系统的处理速率。三、环境准备本文使用以下技术栈Python 3.10 FastAPI Redis redis-py uvicorn安装依赖pip install fastapi uvicorn redis如果本地没有 Redis可以使用 Docker 快速启动docker run -d \ --name redis-limiter \ -p 6379:6379 \ redis:7创建项目目录rate_limit_demo/ ├── main.py ├── redis_client.py ├── limiter.py └── requirements.txt四、连接 Redis先创建redis_client.pyimport redis redis_client redis.Redis( hostlocalhost, port6379, db0, decode_responsesTrue )这里使用同步 Redis 客户端方便演示。在高并发异步项目中也可以使用redis.asyncio。五、实现固定窗口限流固定窗口限流的核心思路是每个限流对象生成一个 Redis key 每次请求让 key 对应的计数器加 1 如果是第一次访问设置过期时间 如果计数超过阈值则拒绝请求例如 key 可以设计为rate_limit:login:127.0.0.1表示某个 IP 对登录接口的访问次数。创建limiter.pyimport time from fastapi import HTTPException, Request from redis_client import redis_client def fixed_window_limiter( key: str, limit: int, window_seconds: int ): 固定窗口限流 key: 限流 key limit: 窗口内最大请求次数 window_seconds: 窗口长度单位秒 current redis_client.incr(key) if current 1: redis_client.expire(key, window_seconds) if current limit: raise HTTPException( status_code429, detailToo Many Requests ) return True这个实现非常简单但有一个细节需要注意current redis_client.incr(key)INCR是 Redis 原子操作因此并发场景下不会出现多个请求同时读取相同旧值的问题。六、在 FastAPI 接口中使用固定窗口限流创建main.pyfrom fastapi import FastAPI, Request from limiter import fixed_window_limiter app FastAPI() app.get(/api/search) def search(request: Request): client_ip request.client.host key frate_limit:search:{client_ip} fixed_window_limiter( keykey, limit5, window_seconds60 ) return { message: search success }启动项目uvicorn main:app --reload访问接口curl http://127.0.0.1:8000/api/search如果同一个 IP 在 60 秒内访问超过 5 次就会返回{ detail: Too Many Requests }状态码为429。七、固定窗口算法的问题固定窗口虽然实现简单但存在明显的边界问题。假设限制规则是每分钟最多请求 5 次用户在第 59 秒请求 5 次然后在第 60 秒之后马上又请求 5 次。从 Redis 计数窗口来看这 10 次请求分别属于两个周期因此都可能被允许。但从实际情况看用户在极短时间内完成了 10 次请求仍然可能对服务造成压力。这就是固定窗口算法的典型缺陷。为了解决这个问题可以使用滑动窗口算法。八、用 Redis ZSET 实现滑动窗口限流滑动窗口需要记录每次请求的时间戳。Redis 的有序集合 ZSET 很适合这个场景。可以把请求时间戳作为 score把请求唯一标识作为 member。每次请求时执行以下步骤1. 删除窗口之前的请求记录 2. 统计当前窗口内的请求数量 3. 如果数量达到限制则拒绝请求 4. 如果没有达到限制则写入当前请求记录 5. 设置 key 过期时间例如限制任意 60 秒内最多请求 5 次当前时间是now则需要保留now - 60 秒 到 now这段时间内的请求记录。代码如下import time import uuid from fastapi import HTTPException from redis_client import redis_client def sliding_window_limiter( key: str, limit: int, window_seconds: int ): 滑动窗口限流 key: 限流 key limit: 窗口内最大请求次数 window_seconds: 窗口长度单位秒 now time.time() window_start now - window_seconds redis_client.zremrangebyscore(key, 0, window_start) current_count redis_client.zcard(key) if current_count limit: raise HTTPException( status_code429, detailToo Many Requests ) member f{now}:{uuid.uuid4()} redis_client.zadd(key, {member: now}) redis_client.expire(key, window_seconds) return True这里使用了几个 Redis 命令ZREMRANGEBYSCORE删除指定分数范围内的元素 ZCARD统计集合中元素数量 ZADD添加元素 EXPIRE设置过期时间每一次请求都会清理过期数据然后检查最近窗口内的请求数量。九、滑动窗口的并发问题上面的滑动窗口实现逻辑是正确的但在极高并发场景下仍然存在一个问题删除旧数据 统计数量 写入新请求这三个步骤不是一个原子操作。如果多个请求同时进来可能都在ZCARD时看到未超限然后一起写入导致短时间内超出限制。为了让操作原子化可以使用 Redis Lua 脚本。十、使用 Lua 脚本保证原子性Redis 执行 Lua 脚本时会把脚本作为一个整体执行。在脚本执行期间不会被其他命令插入。因此适合处理限流这种多步骤操作。Lua 脚本如下SLIDING_WINDOW_LUA local key KEYS[1] local now tonumber(ARGV[1]) local window tonumber(ARGV[2]) local limit tonumber(ARGV[3]) local member ARGV[4] local window_start now - window redis.call(ZREMRANGEBYSCORE, key, 0, window_start) local current redis.call(ZCARD, key) if current limit then return 0 end redis.call(ZADD, key, now, member) redis.call(EXPIRE, key, window) return 1 封装成 Python 函数import time import uuid from fastapi import HTTPException from redis_client import redis_client SLIDING_WINDOW_LUA local key KEYS[1] local now tonumber(ARGV[1]) local window tonumber(ARGV[2]) local limit tonumber(ARGV[3]) local member ARGV[4] local window_start now - window redis.call(ZREMRANGEBYSCORE, key, 0, window_start) local current redis.call(ZCARD, key) if current limit then return 0 end redis.call(ZADD, key, now, member) redis.call(EXPIRE, key, window) return 1 def sliding_window_limiter_lua( key: str, limit: int, window_seconds: int ): now time.time() member f{now}:{uuid.uuid4()} allowed redis_client.eval( SLIDING_WINDOW_LUA, 1, key, now, window_seconds, limit, member ) if allowed ! 1: raise HTTPException( status_code429, detailToo Many Requests ) return True这样清理、统计、判断、写入就变成了一个原子操作。十一、封装成 FastAPI 依赖在实际项目中不建议每个接口都手动写限流逻辑。可以将限流器封装成 FastAPI 依赖。from fastapi import Request def rate_limit( limit: int, window_seconds: int, prefix: str ): def dependency(request: Request): client_ip request.client.host path request.url.path key frate_limit:{prefix}:{path}:{client_ip} sliding_window_limiter_lua( keykey, limitlimit, window_secondswindow_seconds ) return True return dependency使用方式如下from fastapi import FastAPI, Depends app FastAPI() app.get(/api/search, dependencies[Depends(rate_limit( limit5, window_seconds60, prefixsearch ))]) def search(): return { message: search success } app.post(/api/login, dependencies[Depends(rate_limit( limit3, window_seconds60, prefixlogin ))]) def login(): return { message: login success }这样不同接口就可以设置不同的限流规则。例如搜索接口每分钟 5 次 登录接口每分钟 3 次 提交接口每分钟 10 次十二、按用户 ID 限流上面的代码是按 IP 限流。但在登录后的业务系统中更常见的是按用户 ID 限流。例如同一个用户每分钟最多提交 10 次订单可以从请求头或认证信息中获取用户 ID。示例def user_rate_limit( limit: int, window_seconds: int, prefix: str ): def dependency(request: Request): user_id request.headers.get(X-User-Id) if not user_id: raise HTTPException( status_code401, detailUnauthorized ) path request.url.path key frate_limit:{prefix}:{path}:user:{user_id} sliding_window_limiter_lua( keykey, limitlimit, window_secondswindow_seconds ) return True return dependency接口中使用app.post(/api/order, dependencies[Depends(user_rate_limit( limit10, window_seconds60, prefixorder ))]) def create_order(): return { message: order created }需要注意的是真实项目中用户 ID 不应该直接相信请求头而应该从认证系统、JWT、Session 或网关传递的可信身份中获取。十三、返回更友好的错误信息默认返回{ detail: Too Many Requests }对于前端来说可能不够友好。可以统一返回结构from fastapi import HTTPException def raise_rate_limit_error(): raise HTTPException( status_code429, detail{ code: RATE_LIMITED, message: 请求过于频繁请稍后再试 } )在限流失败时调用if allowed ! 1: raise_rate_limit_error()这样前端可以根据code做统一处理。十四、增加 Retry-After 响应头HTTP 中可以使用Retry-After告诉客户端多久后可以重试。示例from fastapi import HTTPException def raise_rate_limit_error(retry_after: int): raise HTTPException( status_code429, detail{ code: RATE_LIMITED, message: 请求过于频繁请稍后再试 }, headers{ Retry-After: str(retry_after) } )例如Retry-After: 60表示客户端可以在 60 秒后重试。在更精细的滑动窗口实现中可以根据最早请求时间计算剩余等待时间。十五、限流 Key 的设计限流效果很大程度上取决于 key 的设计。常见 key 维度包括按 IP 限流 rate_limit:ip:{ip} 按用户限流 rate_limit:user:{user_id} 按接口限流 rate_limit:path:{path} 按用户 接口限流 rate_limit:user:{user_id}:path:{path} 按 IP 接口限流 rate_limit:ip:{ip}:path:{path}不同场景适合不同粒度。登录接口适合按 IP 账号共同限流rate_limit:login:ip:{ip} rate_limit:login:account:{account}这样可以防止同一个 IP 爆破多个账号也可以防止多个 IP 爆破同一个账号。下单接口适合按用户 ID 限流rate_limit:order:user:{user_id}公开查询接口适合按 IP 限流rate_limit:query:ip:{ip}第三方开放接口适合按 API Key 限流rate_limit:openapi:apikey:{api_key}十六、真实项目中的注意事项1. 不要只在应用层限流应用层限流很灵活但在高流量场景下请求已经打到了应用服务。如果流量特别大应该在更靠前的位置限流例如网关层 负载均衡层 Nginx API Gateway 服务网格应用层限流更适合做精细化业务控制。2. 注意代理场景下的真实 IP如果服务部署在反向代理后面直接使用request.client.host拿到的可能是代理服务器 IP而不是真实客户端 IP。这种情况下需要读取X-Forwarded-For X-Real-IP但这些请求头也可能被伪造因此必须由可信网关统一注入和清洗。3. Redis 故障时如何处理限流依赖 Redis。如果 Redis 不可用系统需要决定采用哪种策略fail-openRedis 故障时放行请求 fail-closedRedis 故障时拒绝请求大多数普通业务接口更适合 fail-open避免因为 Redis 故障导致主业务全部不可用。但对于登录、支付、风控等敏感接口可以根据业务风险选择更严格策略。4. 避免 Redis key 爆炸如果 key 维度过细例如用户 ID IP 接口 设备 参数可能会产生大量 Redis key。因此 key 设计需要平衡精度和资源消耗。建议设置合理过期时间避免把长参数直接拼进 key必要时对参数做 hash定期监控 Redis key 数量和内存占用。5. 限流不是权限控制限流只能限制访问频率不能替代认证和权限校验。例如限流解决的是“请求太多”的问题 认证解决的是“你是谁”的问题 权限解决的是“你能不能访问”的问题三者不能混为一谈。十七、完整示例代码下面给出一个简化版完整示例。import time import uuid import redis from fastapi import FastAPI, Request, Depends, HTTPException redis_client redis.Redis( hostlocalhost, port6379, db0, decode_responsesTrue ) SLIDING_WINDOW_LUA local key KEYS[1] local now tonumber(ARGV[1]) local window tonumber(ARGV[2]) local limit tonumber(ARGV[3]) local member ARGV[4] local window_start now - window redis.call(ZREMRANGEBYSCORE, key, 0, window_start) local current redis.call(ZCARD, key) if current limit then return 0 end redis.call(ZADD, key, now, member) redis.call(EXPIRE, key, window) return 1 def raise_rate_limit_error(): raise HTTPException( status_code429, detail{ code: RATE_LIMITED, message: 请求过于频繁请稍后再试 } ) def sliding_window_limiter_lua( key: str, limit: int, window_seconds: int ): now time.time() member f{now}:{uuid.uuid4()} allowed redis_client.eval( SLIDING_WINDOW_LUA, 1, key, now, window_seconds, limit, member ) if allowed ! 1: raise_rate_limit_error() return True def rate_limit( limit: int, window_seconds: int, prefix: str ): def dependency(request: Request): client_ip request.client.host path request.url.path key frate_limit:{prefix}:{path}:{client_ip} sliding_window_limiter_lua( keykey, limitlimit, window_secondswindow_seconds ) return True return dependency app FastAPI() app.get(/api/search, dependencies[Depends(rate_limit( limit5, window_seconds60, prefixsearch ))]) def search(): return { message: search success } app.post(/api/submit, dependencies[Depends(rate_limit( limit3, window_seconds60, prefixsubmit ))]) def submit(): return { message: submit success }启动服务uvicorn main:app --reload测试接口curl http://127.0.0.1:8000/api/search连续请求超过限制后会返回 429。十八、总结本文从零实现了一个基于 FastAPI 和 Redis 的接口限流方案。核心流程可以概括为请求进入接口 ↓ 根据 IP、用户 ID 或 API Key 生成限流 key ↓ 在 Redis 中记录请求时间或请求次数 ↓ 判断是否超过限制 ↓ 允许请求或返回 429固定窗口算法实现简单适合基础场景但存在窗口边界问题。滑动窗口算法更加精确可以限制“任意时间段内”的请求次数更适合对流量控制要求较高的接口。在工程实践中限流方案需要结合业务场景选择合适粒度公开接口可以按 IP 限流用户接口可以按用户 ID 限流登录接口可以按 IP 和账号共同限流开放平台接口可以按 API Key 限流高并发核心接口可以放在网关层提前限流。接口限流不是一个复杂功能但它是后端系统稳定性建设中非常重要的一环。一个好的限流设计既能保护系统不被异常流量拖垮也能让正常用户在高峰期获得更稳定的访问体验。
