使用 TokenBucket 与 FastAPI 构建高效 API 请求限流系统

在现代 Web 应用开发中，API 接口作为前后端数据交互的核心通道，其稳定性与安全性直接关系到整个系统的可用性。然而，随着业务规模扩大，API 被恶意调用、爬虫抓取、高频刷接口等问题日益突出。若不加以控制，轻则造成服务器资源浪费，重则导致服务崩溃或数据泄露。因此，实现高效的请求限流机制成为保障系统健壮性的关键一环。

限流的本质是通过控制单位时间内的请求数量，防止系统被突发流量击穿。常见的限流算法有固定窗口、滑动窗口、漏桶和令牌桶（Token Bucket）。其中，令牌桶算法因其灵活性高、允许短时突发流量的特性，成为许多高并发场景下的首选方案。

为什么选择 TokenBucket？

与固定窗口限流相比，令牌桶算法不仅能平滑处理请求，还能容忍一定程度的流量突增。其核心思想是：系统以恒定速率向“桶”中添加令牌，每个请求需消耗一个令牌才能被执行。若桶中无可用令牌，则请求被拒绝或排队。这种机制既能限制平均速率，又不会过度限制短暂的合法高峰请求。

在 Python 生态中，fustundag/tokenbucket 是一个轻量级且高效的令牌桶实现库，专为异步框架如 FastAPI 设计，具备低延迟、高并发支持等优势，非常适合构建高性能限流中间件。

集成 TokenBucket 到 FastAPI

首先，通过 pip 安装依赖：

bash pip install fastapi uvicorn tokenbucket

接下来，创建一个简单的限流中间件。我们定义每个用户每分钟最多允许 60 次请求，即平均每秒 1 次，支持短时突发。

python
from fastapi import FastAPI, Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
from tokenbucket import MemoryStorage, Limiter
import time

app = FastAPI()

使用内存存储，生产环境可替换为 Redis

storage = MemoryStorage()

创建限流器：容量60，每秒填充1个令牌

limiter = Limiter(storage, capacity=60, fill_rate=1)

class RateLimitMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, callnext): # 以客户端 IP 作为标识符（生产环境建议结合用户ID） identifier = request.client.host key = f"ratelimit:{identifier}"

    # 尝试获取令牌
    current_time = time.time()
    allowed = limiter.try_consume(key, 1, current_time)

    if not allowed:
        raise HTTPException(status_code=429, detail="请求过于频繁，请稍后再试")

    response = await call_next(request)
    return response

app.add_middleware(RateLimitMiddleware)

上述代码将限流逻辑封装为中间件，所有进入的请求都会先经过令牌桶校验。若超出配额，则返回 429 Too Many Requests 状态码。

动态限流策略优化

实际业务中，不同用户、接口或角色可能需要差异化限流策略。例如，普通用户每分钟 60 次，VIP 用户可提升至 300 次。我们可以通过扩展中间件实现动态配置：

python
def getratelimit(usertype: str): limits = { "default": {"capacity": 60, "fillrate": 1},
"vip": {"capacity": 300, "fillrate": 5}, } return limits.get(usertype, limits["default"])

在中间件中根据用户身份动态获取限流参数

allowed = limiter.tryconsume( key, 1, currenttime,
capacity=config["capacity"],
fillrate=config["fillrate"]
)

此外，还可结合 Redis 实现分布式限流，确保集群环境下状态一致性。tokenbucket 支持自定义存储后端，只需实现对应的存储接口即可无缝切换。

实践中的注意事项

• 标识粒度：仅用 IP 可能误伤 NAT 用户，建议结合用户 Token 或设备指纹。
• 异常处理：网络抖动或存储故障不应导致服务不可用，需设置降级策略。
• 监控告警：记录限流触发日志，便于分析攻击行为或调整阈值。
• 透明反馈：可通过响应头 X-RateLimit-Limit, X-RateLimit-Remaining 告知客户端剩余额度。

通过合理使用 fustundag/tokenbucket，我们能在不影响用户体验的前提下，有效抵御 API 滥用风险，提升系统整体稳定性与安全性。