请求限流

请求限流中间件基于 Redis 实现 IP 级别的滑动窗口限流，防止恶意请求和 DoS 攻击。默认关闭，可通过环境变量开启。源码位于 src/middleware/rate_limit.py，滑动窗口 Lua 脚本位于 src/core/redis_client.py。

配置项

环境变量	默认值	说明
RATE_LIMIT_ENABLED	False	是否启用限流
RATE_LIMIT_LIMIT	100	单窗口内每个 IP 最大请求数
RATE_LIMIT_WINDOW_SECONDS	60	统计窗口时长（秒）
RATE_LIMIT_KEY_PREFIX	rl:	Redis 计数 key 前缀

生产环境建议

生产环境建议开启限流（RATE_LIMIT_ENABLED=True），根据实际流量调整 RATE_LIMIT_LIMIT 和 RATE_LIMIT_WINDOW_SECONDS。

工作原理

1. 请求进入 → 检查 RATE_LIMIT_ENABLED 开关
2. 获取客户端 IP（支持 X-Forwarded-For 代理头）
3. 使用滑动窗口算法计算当前窗口请求数
4. 超过限制 → 返回 {"code":1, "msg":"请求过于频繁，请 X 秒后再试"}
5. 未超过 → 放行请求

滑动窗口算法

采用滑动窗口计数器，按当前窗口与上一窗口的计数时间加权，消除固定窗口边界处的 2x 突发问题。

中间件实现

# src/middleware/rate_limit.py
# ============================================================
# 限流中间件
# ============================================================
async def rate_limit_middleware(request: Request, call_next):
    """按客户端 IP 做滑动窗口限流，超限直接返回统一 200 + code 响应"""
    if not RATE_LIMIT_ENABLED:
        return await call_next(request)

    # CORS 预检请求不计数，避免浏览器预检占用限流额度
    if request.method == 'OPTIONS':
        return await call_next(request)

    # 直接读 app.state（lifespan 注入），不依赖 contextvar 注入顺序；Redis 未就绪时降级放行
    redis = getattr(request.app.state, 'redis', None)
    if redis is None:
        return await call_next(request)

    client_ip = get_client_ip(request)
    now = time.time()
    window_index = int(now // RATE_LIMIT_WINDOW_SECONDS)
    current_key = f"{RATE_LIMIT_KEY_PREFIX}{client_ip}:{window_index}"
    prev_key = f"{RATE_LIMIT_KEY_PREFIX}{client_ip}:{window_index - 1}"

    # 原子滑动窗口计数（Lua 内完成 incr/expire/get/加权），消除窗口边界 2× 突发。
    # Redis 运行期故障（连接重置/超时/连接池耗尽，此时 redis 对象仍非 None）时降级放行
    # （fail-open），避免 Redis 一挂全站请求失败——仅判 redis is None 覆盖不到运行期异常
    try:
        count = await sliding_window_incr(
            redis, current_key, prev_key, RATE_LIMIT_WINDOW_SECONDS, window_index, now)
    except Exception as e:
        logger.warning(f"限流计数异常，降级放行: {e}")
        return await call_next(request)

    # 直接返回而非抛异常：BaseHTTPMiddleware 内抛异常会绕过全局异常处理器
    if count > RATE_LIMIT_LIMIT:
        try:
            ttl = await redis.ttl(current_key)
        except Exception:
            ttl = RATE_LIMIT_WINDOW_SECONDS
        return JSONResponse(
            status_code=200,
            content={"code": 1, "data": None, "msg": f"请求过于频繁，请 {ttl} 秒后再试", "ok": False},
        )

    return await call_next(request)

Lua 脚本

滑动窗口的核心在 src/core/redis_client.py 中的 Lua 脚本：

_SLIDING_WINDOW_LUA = """
local cur = redis.call('incr', KEYS[1])
if cur == 1 then
    redis.call('expire', KEYS[1], ARGV[1])
end
local prev = tonumber(redis.call('get', KEYS[2]) or '0')
local elapsed = tonumber(ARGV[3]) - tonumber(ARGV[2]) * tonumber(ARGV[1])
local f = elapsed / tonumber(ARGV[1])
if f < 0 then f = 0 elseif f > 1 then f = 1 end
local effective = prev * (1 - f) + cur
return effective
"""

async def sliding_window_incr(redis, current_key, prev_key,
                              window_seconds, window_index, now) -> float:
    """滑动窗口限流计数：返回当前滑动窗口内的加权请求数（近似精确）"""
    return await redis.eval(
        _SLIDING_WINDOW_LUA, 2, current_key, prev_key,
        window_seconds, window_index, now)

算法原理：当前窗口计数 cur + 上一窗口计数 prev 按时间比例加权（prev * (1 - f)），窗口边界处请求数平滑过渡，消除固定窗口在边界整体重置导致的 2x 突发。

降级策略

1. Redis 未连接：直接放行，不阻塞请求
2. Redis 运行期故障：降级放行，记录警告日志
3. 限流计数异常：降级放行，避免全站不可用

fail-open 策略

限流采用 fail-open 策略：Redis 不可用时放行所有请求，优先保证服务可用性。这是安全与可用性的权衡。

中间件位置

限流中间件在 core/app.py 中注册为最外层（CORS 内层）：

# 最外层：限流（被限流的请求不进入 DB 会话与操作日志）
app.middleware("http")(rate_limit_middleware)
# CORS（确保限流响应也带 CORS 头）
register_cors(app)

总结

请求限流模块具备以下特点：

1. 滑动窗口：消除固定窗口边界处的 2x 突发问题
2. IP 维度：按客户端 IP 限流，支持代理头
3. 默认关闭：opt-in 模式，按需开启
4. 降级策略：Redis 故障时 fail-open 放行
5. 最外层拦截：限流请求不进入 DB 会话和操作日志
6. Lua 原子操作：计数和过期设置原子执行，避免并发问题