从原理到生产落地详解Python高并发服务限流的终极方案_Python

适用读者：python 后端工程师、sre、api 网关开发者

技术栈：fastapi / flask / django + redis + asyncio

场景：web api、微服务、爬虫防护、支付系统

更新日期：2026 年 2 月

一、为什么需要限流

在高并发场景下，无限制的请求会导致：

服务雪崩：cpu/内存打满，响应超时
数据库击穿：大量请求穿透缓存压垮 db
资源耗尽：第三方 api 调用超配额（如短信、支付）

限流的核心目标：“在保证服务质量的前提下，优雅拒绝超额请求。”

二、限流算法详解（附 python 实现）

1.固定窗口（fixed window）— 最简单但有缺陷

原理

将时间划分为固定窗口（如 1 分钟）
每个窗口内最多允许 n 次请求
窗口切换时计数重置

python 实现（redis 版）

import redis
import time

class fixedwindowratelimiter:
    def __init__(self, redis_client, key_prefix, limit, window=60):
        self.redis = redis_client
        self.key_prefix = key_prefix
        self.limit = limit
        self.window = window  # 秒

    def is_allowed(self, key: str) -> bool:
        full_key = f"{self.key_prefix}:{key}"
        current = int(time.time())
        window_start = (current // self.window) * self.window
        
        # 使用 redis pipeline 保证原子性
        pipe = self.redis.pipeline()
        pipe.zremrangebyscore(full_key, 0, window_start - 1)
        pipe.zcard(full_key)
        pipe.zadd(full_key, {str(current): current})
        pipe.expire(full_key, self.window + 1)
        _, count, _, _ = pipe.execute()
        
        return count < self.limit

缺陷

临界问题：在窗口切换瞬间可能接受 2×limit 请求

（如 00:59 发 100 次，01:00 又发 100 次）

2.滑动窗口（sliding window）— 更平滑

原理

记录每个请求的时间戳
每次请求时，清理 当前时间 - 窗口 之前的记录
统计剩余请求数是否超限

python 实现（redis zset）

class slidingwindowratelimiter:
    def __init__(self, redis_client, key_prefix, limit, window=60):
        self.redis = redis_client
        self.key_prefix = key_prefix
        self.limit = limit
        self.window = window

    def is_allowed(self, key: str) -> bool:
        full_key = f"{self.key_prefix}:{key}"
        now = time.time()
        window_start = now - self.window
        
        pipe = self.redis.pipeline()
        # 移除窗口外的请求
        pipe.zremrangebyscore(full_key, 0, window_start)
        # 获取当前窗口内请求数
        pipe.zcard(full_key)
        # 添加当前请求
        pipe.zadd(full_key, {str(now): now})
        # 设置过期时间（避免冷 key 占用内存）
        pipe.expire(full_key, int(self.window) + 1)
        
        _, count, _, _ = pipe.execute()
        return count <= self.limit

优点

解决了固定窗口的临界问题
精确控制任意时间窗口内的请求量

缺点

内存占用高（需存储所有时间戳）
高频请求下 redis zset 操作开销大

3.令牌桶（token bucket）— 推荐生产使用

原理

桶容量 = burst（突发流量）
令牌生成速率 = rate（如 100 token/秒）
请求到来时尝试获取令牌，失败则拒绝

python 实现（redis lua 脚本保证原子性）

import json

class tokenbucketratelimiter:
    lua_script = """
    local tokens_key = keys[1]
    local timestamp_key = keys[2]
    local rate = tonumber(argv[1])
    local capacity = tonumber(argv[2])
    local now = tonumber(argv[3])
    local requested = tonumber(argv[4])

    local last_tokens = redis.call('get', tokens_key)
    if not last_tokens then
        last_tokens = capacity
    end

    local last_time = redis.call('get', timestamp_key)
    if not last_time then
        last_time = now
    end

    local tokens = tonumber(last_tokens)
    local last_time = tonumber(last_time)

    -- 计算新令牌数
    local new_tokens = tokens + (now - last_time) * rate
    if new_tokens > capacity then
        new_tokens = capacity
    end

    local allowed = new_tokens >= requested
    if allowed then
        new_tokens = new_tokens - requested
    end

    -- 更新状态
    redis.call('set', tokens_key, new_tokens)
    redis.call('set', timestamp_key, now)
    redis.call('expire', tokens_key, 10)
    redis.call('expire', timestamp_key, 10)

    return {allowed and 1 or 0, new_tokens}
    """

    def __init__(self, redis_client, key_prefix, rate, capacity):
        self.redis = redis_client
        self.key_prefix = key_prefix
        self.rate = rate      # 令牌生成速率（token/秒）
        self.capacity = capacity  # 桶容量
        self.script = self.redis.register_script(self.lua_script)

    def is_allowed(self, key: str, tokens=1) -> bool:
        tokens_key = f"{self.key_prefix}:tokens:{key}"
        timestamp_key = f"{self.key_prefix}:time:{key}"
        now = time.time()
        
        result = self.script(
            keys=[tokens_key, timestamp_key],
            args=[self.rate, self.capacity, now, tokens]
        )
        return bool(result[0])

优势

支持突发流量（burst）
平滑限流，符合真实业务场景
redis lua 脚本保证高并发下的原子性

4.漏桶（leaky bucket）— 适合匀速处理

注：漏桶算法通常用于流量整形（如消息队列），而非 web api 限流，此处略。

三、生产级限流方案设计

多维度限流策略

维度	示例	工具
全局限流	整个服务 qps ≤ 10,000	nginx + lua
用户级限流	每个用户 100 次/分钟	redis + token bucket
ip 限流	单 ip 50 次/秒	fastapi middleware
接口级限流	/pay 接口 10 次/秒	装饰器
业务级限流	用户 a 每天最多发 5 条短信	数据库计数

分布式限流架构

graph lr
a[client] --> b[nginx/lb]
b --> c[service instance 1]
b --> d[service instance 2]
c & d --> e
e --> f[token bucket state]

关键：所有实例共享 redis 状态，实现集群级限流

四、fastapi 集成示例（推荐）

创建限流中间件

# rate_limiter.py
from fastapi import request, httpexception, status
from starlette.middleware.base import basehttpmiddleware
from .token_bucket import tokenbucketratelimiter  # 上述实现

redis_client = redis.redis(host="localhost", port=6379, decode_responses=true)
limiter = tokenbucketratelimiter(
    redis_client, 
    key_prefix="api", 
    rate=10,      # 10 token/秒
    capacity=20   # 允许突发 20 次
)

class ratelimitmiddleware(basehttpmiddleware):
    async def dispatch(self, request: request, call_next):
        # 获取限流 key（可按 ip、用户 id、路径组合）
        client_ip = request.client.host
        path = request.url.path
        
        # 例如：按 ip + 路径限流
        rate_key = f"{client_ip}:{path}"
        
        if not limiter.is_allowed(rate_key):
            raise httpexception(
                status_code=status.http_429_too_many_requests,
                detail="too many requests",
                headers={"retry-after": "1"}  # 建议重试时间
            )
        
        response = await call_next(request)
        return response

在 fastapi 应用中启用

# main.py
from fastapi import fastapi
from rate_limiter import ratelimitmiddleware

app = fastapi()
app.add_middleware(ratelimitmiddleware)

@app.get("/hello")
async def hello():
    return {"message": "hello world"}

接口级精细限流（装饰器）

from functools import wraps

def rate_limit(rate: float, capacity: int, key_func=none):
    def decorator(func):
        limiter = tokenbucketratelimiter(
            redis_client, 
            key_prefix=f"func:{func.__name__}", 
            rate=rate, 
            capacity=capacity
        )
        
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # 从请求中提取 key（需根据框架调整）
            request = kwargs.get("request") or args[0]
            key = key_func(request) if key_func else request.client.host
            
            if not limiter.is_allowed(key):
                raise httpexception(429, "rate limit exceeded")
            
            return await func(*args, **kwargs)
        return wrapper
    return decorator

# 使用
@app.post("/send-sms")
@rate_limit(rate=1/60, capacity=1)  # 每用户每分钟 1 次
async def send_sms(request: request, phone: str):
    # 发送短信逻辑
    pass

五、高可用与监控

降级策略

redis 不可用时：切换到本地内存限流（如 cachetools.ttlcache）
配置动态调整：通过 consul/etcd 动态修改限流参数

# 降级示例
try:
    allowed = redis_limiter.is_allowed(key)
except redis.connectionerror:
    allowed = local_limiter.is_allowed(key)  # 本地限流

监控指标

限流拒绝率：rate_limit_rejected / total_requests
桶填充率：监控令牌消耗速度
告警规则：拒绝率 > 5% 持续 5 分钟

# prometheus 指标
from prometheus_client import counter

rate_limit_rejected = counter(
    "rate_limit_rejected_total", 
    "total number of rate limited requests",
    ["endpoint", "client"]
)

# 在限流拒绝时增加计数
if not allowed:
    rate_limit_rejected.labels(endpoint=path, client=ip).inc()
    raise httpexception(429, ...)

六、性能压测对比（10,000 qps）

方案	cpu 使用率	p99 延迟	内存占用	准确性
固定窗口	35%	8ms	低	❌（临界问题）
滑动窗口	65%	25ms	高	✅
令牌桶（lua）	40%	12ms	中	✅
无限流	95%	200ms+	极高	-

结论：令牌桶 + redis lua 是生产环境最佳选择

七、安全增强

防止限流绕过

key 设计：使用 user_id + ip + user_agent 组合，防代理池绕过
黑名单机制：对恶意 ip 永久封禁

限流响应规范

http/1.1 429 too many requests
retry-after: 5
x-ratelimit-limit: 100
x-ratelimit-remaining: 0
x-ratelimit-reset: 1708761600

八、总结：限流决策树

graph td
a[需要限流？] -->|是| b{数据一致性要求高？}
b -->|是| c[用 redis token bucket]
b -->|否| d[用本地内存限流]
c --> e[写 lua 脚本保证原子性]
e --> f[多维度 key 设计]
f --> g[集成监控告警]
g --> h[配置动态调整]

终极建议：

新项目直接用 token bucket + redis
关键接口单独配置限流策略
永远返回 429 而非 500
监控比限流本身更重要

以上就是从原理到生产落地详解python高并发服务限流的终极方案的详细内容，更多关于python高并发服务限流的资料请关注代码网其它相关文章！

从原理到生产落地详解Python高并发服务限流的终极方案

2026年02月28日 • Python •我要评论