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.py：
import redis
redis_client = redis.Redis(
host="localhost",
port=6379,
db=0,
decode_responses=True
)
这里使用同步 Redis 客户端，方便演示。
在高并发异步项目中，也可以使用 redis.asyncio。
五、实现固定窗口限流
固定窗口限流的核心思路是：
每个限流对象生成一个 Redis key
每次请求让 key 对应的计数器加 1
如果是第一次访问，设置过期时间
如果计数超过阈值，则拒绝请求
例如 key 可以设计为：
rate_limit:login:127.0.0.1
表示某个 IP 对登录接口的访问次数。
创建 limiter.py：
import 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_code=429,
detail="Too Many Requests"
)
return True
这个实现非常简单，但有一个细节需要注意：
current = redis_client.incr(key)
INCR 是 Redis 原子操作，因此并发场景下不会出现多个请求同时读取相同旧值的问题。
六、在 FastAPI 接口中使用固定窗口限流
创建 main.py：
from 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 = f"rate_limit:search:{client_ip}"
fixed_window_limiter(
key=key,
limit=5,
window_seconds=60
)
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_code=429,
detail="Too 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_code=429,
detail="Too 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 = f"rate_limit:{prefix}:{path}:{client_ip}"
sliding_window_limiter_lua(
key=key,
limit=limit,
window_seconds=window_seconds
)
return True
return dependency
使用方式如下：
from fastapi import FastAPI, Depends
app = FastAPI()
@app.get("/api/search", dependencies=[Depends(rate_limit(
limit=5,
window_seconds=60,
prefix="search"
))])
def search():
return {
"message": "search success"
}
@app.post("/api/login", dependencies=[Depends(rate_limit(
limit=3,
window_seconds=60,
prefix="login"
))])
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_code=401,
detail="Unauthorized"
)
path = request.url.path
key = f"rate_limit:{prefix}:{path}:user:{user_id}"
sliding_window_limiter_lua(
key=key,
limit=limit,
window_seconds=window_seconds
)
return True
return dependency
接口中使用：
@app.post("/api/order", dependencies=[Depends(user_rate_limit(
limit=10,
window_seconds=60,
prefix="order"
))])
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_code=429,
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_code=429,
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-open：Redis 故障时放行请求
fail-closed：Redis 故障时拒绝请求
大多数普通业务接口更适合 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(
host="localhost",
port=6379,
db=0,
decode_responses=True
)
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_code=429,
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 = f"rate_limit:{prefix}:{path}:{client_ip}"
sliding_window_limiter_lua(
key=key,
limit=limit,
window_seconds=window_seconds
)
return True
return dependency
app = FastAPI()
@app.get("/api/search", dependencies=[Depends(rate_limit(
limit=5,
window_seconds=60,
prefix="search"
))])
def search():
return {
"message": "search success"
}
@app.post("/api/submit", dependencies=[Depends(rate_limit(
limit=3,
window_seconds=60,
prefix="submit"
))])
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 限流；
-
高并发核心接口可以放在网关层提前限流。
接口限流不是一个复杂功能，但它是后端系统稳定性建设中非常重要的一环。
一个好的限流设计，既能保护系统不被异常流量拖垮，也能让正常用户在高峰期获得更稳定的访问体验。
汇聚全球AI编程工具，助力开发者即刻编程。
更多推荐
9
0- 0
已为社区贡献1条内容
所有评论(0)