限流不是把用户挡在外面：django-ratelimit、令牌桶与配额系统的实战

起因是一个很常见的场景：我们的解析接口被某几个 IP 刷了，一天 40 多万次请求，带宽和第三方解析额度直接烧掉大半。

第一版修复很粗暴：@ratelimit(key='ip', rate='60/m', block=True)。上线两小时后收到投诉：一整个公司的用户都访问不了。查下来是他们在 NAT 后面，几百个人共用一个出口 IP，60 次/分钟根本不够用。

那次之后我把限流重做了一遍，分成了三层：速率限制（防刷）、并发闸门（保护下游）、配额（按用户等级的公平使用）。这篇写完整的设计、代码，以及那次误伤事故带来的教训。

TL;DR：限流有三个维度，别混为一谈——速率（每秒多少次，挡爬虫）、并发（同时多少个，保护下游服务）、配额（每天总量，做商业分级）。算法固定窗口/滑动窗口够简单，但要允许突发就用令牌桶（Redis + Lua 原子实现）。关键经验：别只按 IP 限（NAT 会误伤一片），登录用户按用户 ID 限、匿名用户按 IP 且放宽，配合白名单和 Retry-After 头。命中限流返回 429 并在响应里告诉客户端多久后重试。

一、限流到底在限什么：三个维度
二、四种限流算法与选择
三、令牌桶的 Redis + Lua 实现（可复制）
四、Django 层：django-ratelimit 的正确用法
五、nginx 层：limit_req 与 limit_conn
六、配额系统：按用户等级的公平使用
七、并发闸门：保护下游不被自己打垮
八、防误伤：那次 NAT 事故换来的六条规则
九、可观测：限流也要有监控
十、对抗：换 IP、分布式刷量怎么办
十一、坑清单

一、限流到底在限什么：三个维度

很多人说"加个限流"，其实指的是三种完全不同的东西：

维度	限制什么	目的	典型配置
速率（rate）	每秒/分钟请求数	防刷、防爬虫	60 次/分钟
并发（concurrency）	同时在处理的请求数	保护下游服务不被打垮	下游同时最多 8 个
配额（quota）	一段时间内的总量	商业分级、成本控制	免费用户每天 20 次

三个要同时存在。只有速率限制的话，一个用户可以把额度在 1 秒内用完（对后端是突发冲击）；只有配额的话，一个用户可以瞬间发 1000 个并发请求把你打垮。

我们的最终结构：

用户请求
   ↓
[nginx]  limit_req（IP 级粗限流，挡最粗暴的刷量）
   ↓
[Django] 认证 → 配额检查（用户每日额度）
   ↓
[Django] 速率限制（用户/令牌桶，允许突发）
   ↓
[并发闸门] 下游解析服务的同时请求数上限
   ↓
业务处理

二、四种限流算法与选择

算法	原理	优点	缺点	适合
计数器（固定窗口）	每分钟计数，超了拒绝	最简单	临界突刺：59 秒和 61 秒各 100 次 = 2 秒内 200 次	粗粒度挡刷
滑动窗口	统计最近 60 秒	无临界问题	要存每次请求的时间戳，占内存	精确限流
令牌桶	桶里按速率生成令牌，取到才能过	允许突发（桶容量）、平滑	实现稍复杂	推荐
漏桶	恒定速率出水	绝对平滑	不允许任何突发	流量整形（下游）

我的选择：

挡爬虫/粗粒度：固定窗口就够（nginx 的 limit_req 其实是漏桶）；
给正常用户的接口限流：令牌桶，因为它允许突发——用户点了几下不会立刻被拒，但持续高频会被限。

令牌桶的直观理解：

桶容量 = 20（允许瞬间来 20 个请求）
生成速率 = 5 个/秒（持续速率是 5 QPS）

用户 1 秒内来了 20 个请求 → 桶里有 20 个令牌，全部通过
第 21 个 → 拒绝（429）
1 秒后 → 桶里补了 5 个，又能过 5 个

这就是为什么令牌桶比固定窗口体验好：正常用户的行为是"点一下、看一会儿、再点一下"，桶里攒着的令牌正好覆盖这种突发；而脚本的持续高频会被平均速率卡住。

三、令牌桶的 Redis + Lua 实现（可复制）

为什么用 Lua：限流需要"读-算-写"三步，分开做有竞态（两个请求同时读到 10 个令牌，都通过）。Lua 脚本在 Redis 里原子执行。

import time
from django.core.cache import cache

# 需要在 django-redis 下拿到原生 redis 客户端
def _redis():
    return cache.client.get_client(write=True)

TOKEN_BUCKET_LUA = """
local key = KEYS[1]
local rate = tonumber(ARGV[1])        -- 每秒生成的令牌数
local burst = tonumber(ARGV[2])       -- 桶容量（允许的突发量）
local now = tonumber(ARGV[3])         -- 当前时间戳（秒，可含小数）
local requested = tonumber(ARGV[4])   -- 本次要取的令牌数

local ttl = math.ceil(burst / rate) + 1
local last = redis.call('HMGET', key, 'ts', 'tokens')
local last_ts = tonumber(last[1])
local tokens = tonumber(last[2])

if last_ts == nil then
    last_ts = now
    tokens = burst
end

-- 按时间差补充令牌，不超过桶容量
local delta = math.max(0, now - last_ts)
local filled = math.min(burst, tokens + delta * rate)

local allowed = 0
if filled >= requested then
    allowed = 1
    filled = filled - requested
end

redis.call('HSET', key, 'ts', now, 'tokens', filled)
redis.call('EXPIRE', key, ttl)
return {allowed, math.floor(filled)}
"""

_script = None


def token_bucket(key: str, rate: float, burst: int, requested: int = 1):
    """令牌桶限流。返回 (是否放行, 剩余令牌数)。"""
    global _script
    if _script is None:
        _script = _redis().register_script(TOKEN_BUCKET_LUA)
    allowed, remaining = _script(keys=[key], args=[rate, burst, time.time(), requested])
    return bool(int(allowed)), int(remaining)


def retry_after(key: str, rate: float) -> int:
    """估算还要多久才有令牌（用于 Retry-After 头）。"""
    data = _redis().hgetall(key)
    tokens = float(data.get(b'tokens') or 0)
    need = 1 - tokens
    return max(1, int(need / rate) + 1) if need > 0 else 1

使用：

from django.http import JsonResponse

def parse_view(request):
    user = request.user
    key = f'rl:parse:user:{user.id}' if user.is_authenticated else f'rl:parse:ip:{get_ip(request)}'
    allowed, remaining = token_bucket(key, rate=1.0, burst=10)   # 持续 1 QPS，突发 10

    if not allowed:
        resp = JsonResponse({'error': '请求过于频繁，请稍后再试'}, status=429)
        resp['Retry-After'] = str(retry_after(key, 1.0))
        resp['X-RateLimit-Remaining'] = '0'
        return resp

    resp = do_parse(request)
    resp['X-RateLimit-Remaining'] = str(remaining)
    return resp

Retry-After 头很重要：它告诉客户端（和用户的浏览器/脚本）多久后可以重试。有了它，正常用户的客户端会自动退避重试，体验好很多；没有它，用户只能不停手动刷新（反而加重负担）。

四、Django 层：django-ratelimit 的正确用法

django-ratelimit（requirements 里有 django-ratelimit==4.1.0）提供了最简单的接入：

from django_ratelimit.decorators import ratelimit

@ratelimit(key='user', rate='100/h', method='POST', block=True)
def parse(request):
    ...

key 的选择是关键：

key	含义	什么时候用
`'ip'`	客户端 IP	匿名接口（但注意 NAT）
`'user'`	`request.user.pk`	登录后按用户限（推荐）
`'user_or_ip'`	登录用 user，否则用 ip	混合接口
`'header:x-api-key'`	自定义头	API 用户
callable	自己算	复杂场景

我们的用法：

from django_ratelimit.decorators import ratelimit

# 登录接口：按 IP 限（防暴力破解），严格
@ratelimit(key='ip', rate='10/m', method='POST', block=True)
def login_view(request):
    ...

# 解析接口：登录用户按用户限，宽松；匿名按 IP，中等
@ratelimit(key='user_or_ip', rate='30/m', method='POST', block=True)
def parse_view(request):
    ...

# 敏感操作：非常严格
@ratelimit(key='user', rate='5/h', method='POST', block=True)
def send_sms(request):
    ...

自定义 key（比如区分用户等级）：

def rate_key(group, request):
    if request.user.is_authenticated:
        return f'{group}:u{request.user.id}'
    return f'{group}:ip{request.META.get("REMOTE_ADDR")}'

@ratelimit(key=rate_key, rate='60/m', method='POST', block=True)
def some_view(request):
    ...

被限流时默认返回 403。想返回 429 的话（更符合语义）：

from django_ratelimit.exceptions import Ratelimited
from django.http import JsonResponse


class RatelimitMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        return self.get_response(request)

    def process_exception(self, request, exception):
        if isinstance(exception, Ratelimited):
            return JsonResponse(
                {'error': '请求过于频繁', 'retry_after': 60},
                status=429,
                headers={'Retry-After': '60'},
            )
        return None

然后配置 RATELIMIT_VIEW? 不，django-ratelimit 4.x 通过 middleware 捕获 Ratelimited 异常（block=True 时抛出）。

五、nginx 层：limit_req 与 limit_conn

nginx 层的限流是第一道防线，它挡在最前面，不消耗 Python 资源：

http {
    # 按 IP 限请求速率：10 r/s，允许突发 20
    limit_req_zone $binary_remote_addr zone=api_rate:10m rate=10r/s;

    # 按 IP 限并发连接
    limit_conn_zone $binary_remote_addr zone=api_conn:10m;

    server {
        location /api/ {
            limit_req zone=api_rate burst=20 nodelay;
            limit_conn api_conn 10;
            limit_req_status 429;        # 默认 503，改成 429 更准确
            limit_conn_status 429;

            proxy_pass http://django;
            ...
        }
    }
}

几个参数的含义：

参数	含义
`rate=10r/s`	平均速率上限
`burst=20`	允许突发 20 个排队
`nodelay`	突发的不排队而是立即处理（不延迟响应）；不加则排队处理（更平滑但更慢）
`limit_req_status 429`	被限时的状态码

nodelay 的选择：

加了：突发请求立即处理，超出直接 429（响应快，体验干脆）；
不加：突发请求排队，慢慢处理（延迟高但不会拒绝）。

对外 API 我一般加 nodelay——用户宁愿快速失败重试，也不愿意等 3 秒。

注意 limit_req_zone 要写在 http 块里，它定义的是"共享内存区"，不能写在 server 里。

限流的粒度：$binary_remote_addr 是二进制格式的 IP（省内存）。也可以按别的维度：

# 按用户 ID（从 cookie 里取）
limit_req_zone $cookie_userid zone=user_rate:10m rate=30r/m;

# 按 Authorization 头
limit_req_zone $http_authorization zone=token_rate:10m rate=100r/h;

六、配额系统：按用户等级的公平使用

速率限制防的是"短时间刷量"，配额管的是"总量"。我们的会员分级就靠它。

from django.core.cache import cache
from django.utils import timezone

QUOTAS = {
    'free':    20,      # 每天 20 次
    'basic':   200,
    'pro':     2000,
    'staff':   10 ** 9,
}


def _quota_key(user_id: int) -> str:
    today = timezone.localdate().isoformat()
    return f'quota:{today}:u{user_id}'


def quota_left(user) -> int:
    """剩余配额。跨天自动重置（key 里带日期）。"""
    used = int(cache.get(_quota_key(user.id)) or 0)
    total = QUOTAS.get(getattr(user, 'level', 'free'), QUOTAS['free'])
    return max(0, total - used)


def consume_quota(user, n: int = 1) -> bool:
    """扣减配额。返回是否成功。"""
    key = _quota_key(user.id)
    total = QUOTAS.get(getattr(user, 'level', 'free'), QUOTAS['free'])

    try:
        new_val = cache.incr(key, n)
    except ValueError:
        # key 不存在，incr 会抛 ValueError
        cache.set(key, n, timeout=86400 + 3600)      # 跨天多留 1 小时
        new_val = n

    if new_val > total:
        cache.decr(key, n)          # 超了要还回去
        return False
    return True

几个设计点：

按日期做 key，自动跨天重置，不用写定时任务去清零；
超时时间设为 25 小时（跨天多留一点），避免时区边缘出错；
扣减失败要回滚（decr），否则"额度不足"的失败请求也会把额度吃掉；
cache.incr 是原子的，并发安全。

配额数据库对账：Redis 会丢（重启、内存淘汰）。重要场景要把配额落库：

# 任务成功后落库，每天对账一次
class UsageRecord(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    date = models.DateField(auto_now_add=True)
    count = models.PositiveIntegerField(default=0)
    class Meta:
        unique_together = ('user', 'date')

@shared_task
def reconcile_quota():
    """把 Redis 里的计数同步进数据库（每天一次），并修正漂移。"""
    ...

这是必须的——纯 Redis 计数一旦丢失，用户可以无限用；而如果计数异常偏大，用户会被无故限流。我们每周一早上跑一次对账，同时输出"Redis 计数 vs 数据库计数"的差值告警。

七、并发闸门：保护下游不被自己打垮

我们的解析服务要调用第三方接口，对方有并发限制（同时最多 5 个）。如果不限制，8 个 gunicorn worker × 12 = 一大堆并发请求打过去，对方直接封 IP。

用 Redis 实现的分布式信号量：

import time
import uuid
from django.core.cache import cache


class ConcurrencyGate:
    """简单的分布式并发闸门（基于 Redis 有序集合）。"""

    def __init__(self, name: str, limit: int, timeout: int = 30):
        self.key = f'gate:{name}'
        self.limit = limit
        self.timeout = timeout

    def acquire(self) -> str | None:
        now = time.time()
        token = uuid.uuid4().hex
        pipe = cache.client.get_client(write=True).pipeline()
        # 清理超时的持有者
        pipe.zremrangebyscore(self.key, '-inf', now - self.timeout)
        pipe.zadd(self.key, {token: now})
        pipe.zcard(self.key)
        _, _, count = pipe.execute()

        if count > self.limit:
            # 超了，把自己撤回来
            cache.client.get_client(write=True).zrem(self.key, token)
            return None
        return token

    def release(self, token: str):
        cache.client.get_client(write=True).zrem(self.key, token)

用法：

gate = ConcurrencyGate('thirdparty_parse', limit=5, timeout=60)

def call_third_party(url):
    token = None
    for _ in range(30):            # 最多等 3 秒
        token = gate.acquire()
        if token:
            break
        time.sleep(0.1)
    if not token:
        raise ServiceBusy('服务繁忙，请稍后再试')
    try:
        return requests.get(url, timeout=10)
    finally:
        gate.release(token)

timeout 是保险：万一某个请求异常退出没释放，超时后自动被清理，不会永久占着位置。

也可以用现成的库（redis-semaphore、django-redis 的 lock），但自己实现 30 行就够，而且行为完全可控。

八、防误伤：那次 NAT 事故换来的六条规则

那次把一个公司两百人挡在门外之后，我们定了六条规则：

1. 登录用户按用户 ID 限，IP 限制只作为兜底

def rate_key(group, request):
    if request.user.is_authenticated:
        return f'{group}:u{request.user.id}'
    return f'{group}:ip{request.META.get("REMOTE_ADDR")}'

2. 匿名用户的 IP 限制要宽松（我们设的是登录用户的 2~3 倍），因为一个 IP 后面可能是很多人。

3. 白名单机制

WHITELIST_IPS = {'203.0.113.0/24'}     # 客户公司出口
WHITELIST_USERS = {1, 2, 3}            # 内部账号


def is_whitelisted(request) -> bool:
    ip = get_ip(request)
    if any(ipaddress.ip_address(ip) in ipaddress.ip_network(n) for n in WHITELIST_IPS):
        return True
    return request.user.is_authenticated and request.user.id in WHITELIST_USERS

4. 内网的监控/健康检查不受限（否则告警系统自己先被限流，你就瞎了）。

5. 被限流返回 429 + Retry-After + 明确的错误信息，前端要友好提示"稍后再试"，而不是弹一个看不懂的错误码。

6. 命中限流时记录日志并告警，如果某个 IP 或用户频繁被限，要么是真被刷（正常），要么是我们的阈值设太低（要调整）。这条是发现误伤的关键——那次事故如果有限流日志，我两分钟就能定位，而不是等用户投诉。

还有一条限流的"灰度"做法：新规则先只记录不拦截（block=False），观察一周的命中情况，确认没有大面积误伤之后再开启拦截。django-ratelimit 支持：

@ratelimit(key='user_or_ip', rate='30/m', method='POST', block=False)
def parse_view(request):
    if getattr(request, 'limited', False):
        logger.warning('would be limited: %s', ...)
    ...

强烈建议所有新限流规则先灰度一周。

九、可观测：限流也要有监控

限流不是配完就完事的，要看数据：

# 统一的限流命中埋点
import logging
from django.core.cache import cache

logger = logging.getLogger('ratelimit')


def record_limit(kind: str, subject: str):
    logger.warning('ratelimit hit', extra={'kind': kind, 'subject': subject})
    # 按天计数，看趋势
    key = f'stats:ratelimit:{timezone.localdate().isoformat()}:{kind}'
    try:
        cache.incr(key)
    except ValueError:
        cache.set(key, 1, timeout=7 * 86400)

监控指标：

指标	告警阈值	含义
限流命中总数（每小时）	突然涨 10 倍	可能正在被刷
被限的不同 IP 数	突然涨	分布式刷量
某个用户被限次数	> 50 次/天	可能是误伤，要查
配额耗尽的用户数	趋势上涨	可能是配额太低或业务增长

"某个用户被限次数过多"这条最有价值——它直接指向误伤。我们现在的告警是"单个用户一天被限超过 50 次就发群消息"，基本每次都能及时发现配置问题。

十、对抗：换 IP、分布式刷量怎么办

限流能挡住的是"不那么专业的刷量"。遇到换 IP 的：

逐级升级的应对（成本从低到高）：

IP 段限制：不再按单个 IP，按 /24 段统计（ipaddress.ip_network(f'{ip}/24', strict=False)）。能挡住同一段内的 IP 轮换。
User-Agent + 指纹：正常浏览器的 UA 是有限的几种，脚本往往用默认 UA 或者固定 UA。配合 TLS 指纹（JA3）效果更好（项目里那篇反检测的文章讲过）。
行为特征：请求间隔完全均匀（比如精确 1 秒一个）几乎一定是脚本；正常用户是随机的。
验证码：命中可疑行为后要求人机验证。
阶梯封禁：第一次警告、第二次限流加倍、第三次临时封禁。

我一般的策略是先做 1 和 3，因为它们零成本且无副作用。验证码是最后手段——它对所有用户的体验都有损害，别轻易上。

还有一条很重要的：不要试图做到 100% 防住。 专业的对抗是没有终点的（他们换 IP 的成本比你的防御成本低）。合理的目标是把刷量的成本提高到"不值得"，而不是彻底消灭。

十一、坑清单

只按 IP 限流 → NAT 后面的公司/学校用户全被误伤。登录用户按用户 ID 限。
限流规则直接上 block=True → 大面积误伤。先 block=False 灰度观察一周。
不返回 Retry-After → 用户/客户端不知道多久能重试，只能不停刷新，雪上加霜。
返回 403 而不是 429 → 语义不对，客户端无法区分"被限流"和"没权限"。
只有速率限制没有并发限制 → 突发请求打垮下游。
只有配额没有速率限制 → 一个用户一秒内用完一个月额度。
配额用 Redis 计数但不对账 → Redis 重启后额度清零（用户白嫖），或者计数漂移（用户被无故限制）。
扣减配额失败没回滚 → 失败的请求也吃掉了额度。
固定窗口的临界突刺 → 窗口切换处能打出两倍流量。要精确就用滑动窗口或令牌桶。
限流 key 冲突（比如用用户名而不是 ID，改名就失效）→ 用不可变的 ID。
limit_req_zone 写在了 server 块里 → nginx 报错，配置不生效。
健康检查接口也被限流 → 告警系统自己先挂，事故发现得更晚。
并发闸门没有超时 → 异常退出的请求永久占着位置，闸门慢慢锁死。
限流命中不记日志 → 出事了不知道是限流导致的，也不知道有没有误伤。
为了防刷上了验证码 → 所有用户体验下降。先做零成本的行为分析和 IP 段限制。

最后说说这次重做限流的整体感受。

限流的难点从来不是技术实现（令牌桶三十行代码，nginx 两行配置），而是阈值定在哪里。定高了挡不住，定低了误伤正常用户。而这个问题没有标准答案，只能靠数据：

先灰度一周，看正常用户的实际请求分布（P99 是多少）；
阈值设在 P99 的 2~3 倍（给正常用户留足余量）；
上线后持续看"被限的用户数"和"单个用户被限次数"，发现异常立刻调整。

我们现在的解析接口限制是 30 次/分钟，这个数字不是拍脑袋的——灰度期间观察到正常用户的 P99 是 8 次/分钟，我们取了接近 4 倍。阈值要有依据，而且要可调整（我们把它放在了配置表里，改了不用发版）。

还有一点：限流是"对所有人都不信任"的机制，但它不该让正常用户感觉到存在。如果某个限流规则让正常用户频繁看到"请求过于频繁"，那这个规则就是错的，不管它挡住了多少恶意流量。用户投诉的成本，通常比多花的那点带宽贵得多。