客户端限流怎么做：与其撞 429，不如自己先排队

2026-08-07

一个很常见的现场：业务量涨了一波，日志里开始刷 429，有人第一时间去加重试次数，从 3 次改成 5 次，退避从 1 秒起步改成 2 秒起步。改完当天确实”好了”——错误率降下去了。一周后同样的位置再炸，这次退避改到 8 次，延迟曲线开始难看，成本账单也没见少。

问题出在方向上。撞了 429 再重试，是被上游拒绝之后的补救；而真正该做的事情发生在更早的位置——这次请求根本不应该被发出去。

我见过不少团队跳过客户端限流这一步。理由通常是”上游本来就有限速，我打过去它会告诉我”。这个理由在功能上说得通，在经济上完全不成立。

被拒绝的请求不是免费的

一个被上游用 429 打回来的请求，你以为它什么都没花。实际上它花掉了这些东西：

一次完整的网络往返。跨境链路上这一趟本身就不便宜，而且它占用了你这边的连接、线程或协程、以及超时预算里的一段时间。
上游的计数。多数平台的限速计数发生在你被拒绝之前而不是之后，也就是说，被拒的请求同样算你头上。你越是拼命重发，越是在往自己的配额上撞。
可能更严的处置。持续高频撞限的调用方，在很多平台的风控视角里跟异常流量长得一模一样。具体每家怎么处置以官方文档为准，但”越撞越顺”这种情况是不存在的。
你自己的错误率指标。429 是四百开头的，很多团队的监控面板会把它算进接口失败率，导致告警噪音，值班的人被吵醒之后发现”哦是限流”，下次真出事的时候他就不看了。

而客户端排队呢？排队只花时间，不花往返、不花配额、不花上游的耐心。同样是”这个请求现在过不去”，在自己这边等，和被别人打回来,代价完全不是一个量级。

这就是客户端限流的全部经济学理由：把”过不去”这件事的成本，从”往返 + 配额 + 风险”压到只剩”时间”。

先搞清楚上游的限速模型长什么样

在写任何限流代码之前，先花半小时把要接的这家平台的限速文档读完。这一步跳过去的话，后面写的限流器大概率是在防一个不存在的维度。

限速模型有两个特征，几乎每次都会让第一次接的人翻车。

特征一：多维度同时生效，先撞到哪个算哪个

限速很少只有”每分钟多少次请求”这一个口径。常见的维度至少有四个：

维度	常见写法	约束的是什么
每分钟请求数	RPM	你打出去的调用频次
每分钟 token 数	TPM	你搬运的数据量
每天请求数	RPD	一天总的调用次数
每天 token 数	TPD	一天总的数据量

以 Groq 为例，其官方文档明确说明：先撞到哪个阈值，哪个阈值就生效（来源：Groq 官方 rate limits 文档）。也就是说这几条限制是”或”的关系，不是”与”的关系——你只要触碰到其中任意一条，请求就会被拒。

这条机制的实际后果很反直觉。假设你的 RPM 用量只到配额的三成，看面板一片绿，你完全可能同时在 TPM 上撞得头破血流——因为你的请求虽然不多，但每个都带着很长的上下文。反过来也成立：一堆短请求可以在 token 维度上很闲，却把 RPM 打满。

只盯一个维度做限流，等于给自己留了三扇没关的门。

特征二：限速常常是组织级的，不是用户级的

第二条更容易出事。Groq 官方文档里写得很直白：速率限制适用于组织级别，不适用于个别用户（来源同上）。

各家平台的口径不一样，有的按 API key 算，有的按项目算，有的按组织账户算——具体机制以各平台官方文档为准。但只要遇到”按组织算”这种口径，一件事就成立了：

你的开发环境、CI 流水线、预发环境、生产环境，如果挂在同一个组织下，它们在抢同一个池子。

我见过最典型的一次翻车，是 CI 里加了一组回归测试，跑起来是并发的，跑一次几百个调用。这套测试在工作时间跟着每个 PR 跑，然后生产环境在同一时段开始零星报 429。查了两天，因为所有人都在看生产的代码，没人想到问题出在 CI。

隔离的做法，按代价从低到高：

不同环境用不同的 key。这是最低成本的一步，很多平台的用量统计能按 key 拆开，至少让你事后能归因。但注意：如果限速是组织级的，换 key 并不能分开配额，它只解决”看得见”，不解决”抢不到”。
不同环境用不同的组织/项目账户。这才是真正把池子分开。代价是账号管理和账单变复杂。
给非生产流量单独设一个更小的客户端配额。哪怕池子是共享的，你也可以在自己这边规定 CI 最多用总配额的 20%，用限流器去执行这条规矩。这是没法拆账号时的次优解。
给请求打标记。在请求里带上环境标识（比如自定义 header 或者 user 字段，具体支持哪些字段以平台文档为准），出事的时候能一眼看出是谁在打。

第 3 条尤其值得强调：限流器不只是用来防上游拒绝的，它也是你在内部分配配额的执行工具。 想清楚这一点，下面的实现选择才有意义。

客户端限流的三种实现

一、信号量：控制同时在途的请求数

最简单的一种，本质是”最多允许 N 个请求同时在飞”。

import asyncio

class ConcurrencyLimiter:
    def __init__(self, max_inflight: int):
        self._sem = asyncio.Semaphore(max_inflight)

    async def run(self, coro_factory):
        async with self._sem:
            return await coro_factory()

适用场景：你只知道”大概能同时压多少”，不知道上游的 RPM/TPM 具体是多少；或者你真正想保护的是自己这边的资源（连接池、内存、下游数据库）。

代价：它控制的是并发度，不是速率。这两个不是一回事。如果每个请求都很快返回，N=10 的信号量可能一分钟打出去几千次调用；如果每个请求要跑 30 秒，同样的 N=10 一分钟只有 20 次。同一个信号量配置，在流式长回答和短分类任务上的实际速率能差两个数量级。

所以信号量适合做兜底和保护自己，不适合用来精确对齐上游的 RPM 配额。并发度这条线怎么定，并发数怎么设那篇讲得更细。

二、令牌桶：控制速率，允许一定突发

令牌桶的模型是：桶按固定速率往里滴令牌，桶有容量上限；每个请求出发前先拿一个令牌，拿不到就等。

它比信号量强在两点：直接对齐”每分钟多少次”这个口径；桶容量决定了允许多大的突发——桶攒满了之后可以一口气放出一批，之后回落到稳定速率。

import time
import threading

class TokenBucket:
    def __init__(self, rate_per_sec: float, capacity: float):
        self.rate = rate_per_sec        # 每秒补充多少令牌
        self.capacity = capacity        # 桶容量，决定允许的突发大小
        self._tokens = capacity
        self._last = time.monotonic()
        self._lock = threading.Lock()

    def _refill(self):
        now = time.monotonic()
        self._tokens = min(self.capacity, self._tokens + (now - self._last) * self.rate)
        self._last = now

    def try_acquire(self, need: float = 1.0) -> bool:
        with self._lock:
            self._refill()
            if self._tokens >= need:
                self._tokens -= need
                return True
            return False

    def wait_time(self, need: float = 1.0) -> float:
        """还要等多久才能拿到 need 个令牌，单位秒。"""
        with self._lock:
            self._refill()
            if self._tokens >= need:
                return 0.0
            return (need - self._tokens) / self.rate

桶容量怎么定，是这里唯一需要动脑子的参数。 容量等于速率乘以 1 秒，基本就是匀速，突发能力约等于零，好处是最不容易撞限，坏处是排队延迟高。容量放到速率乘以 10 秒，突发能力强，业务侧的尖峰能被吸收，但你有可能在那 10 秒里把上游的分钟窗口提前吃掉一大块。

我的经验是：如果上游的窗口是分钟级，桶容量不要超过速率乘以窗口长度的四分之一。 这样即使突发全放出去，也只吃掉这一分钟配额的四分之一，剩下的时间还有余量。这是个工程判断，不是什么定律，但它比”直接设成一分钟的量”要稳得多。

三、双维度限流：RPM 和 TPM 各一个桶，取更严的那个

这是最贴合实际的做法，也是三种里唯一真正对得上前面那张多维表的。

思路很直接：维护两个令牌桶，一个按请求数计，一个按 token 数计。请求发出前，先估算这次要消耗多少 token，然后两个桶都要能通过才放行；任一个不够，就按两者中更长的等待时间去等。

class DualLimiter:
    def __init__(self, rpm: int, tpm: int, burst_ratio: float = 0.25):
        # 速率换算成每秒；容量取窗口的一部分作为允许的突发
        self.req_bucket = TokenBucket(rpm / 60.0, capacity=rpm * burst_ratio)
        self.tok_bucket = TokenBucket(tpm / 60.0, capacity=tpm * burst_ratio)

    def acquire(self, est_tokens: float, timeout: float | None = None) -> bool:
        deadline = None if timeout is None else time.monotonic() + timeout
        while True:
            w = max(self.req_bucket.wait_time(1.0),
                    self.tok_bucket.wait_time(est_tokens))
            if w == 0.0:
                # 两个桶同时扣，避免只扣一个造成计数漂移
                if self.req_bucket.try_acquire(1.0) and self.tok_bucket.try_acquire(est_tokens):
                    return True
                continue  # 并发争抢导致的罕见失败，重新计算
            if deadline is not None and time.monotonic() + w > deadline:
                return False   # 等不起，交给上层决定拒绝还是降级
            time.sleep(min(w, 0.5))

token 数怎么估？ 关键是”发送前”就得有个数，不能等响应回来才知道。可用的办法按精度从高到低：

用官方 tokenizer 离线算输入侧。这是最准的，输入部分能算到几乎精确。代价是引入一个依赖，而且不同模型的 tokenizer 不通用。
用字符数做经验换算。中英文混排的换算比例差别很大，得按你自己的真实语料标定一次，别抄别人的系数——同一段中文在不同模型的 tokenizer 下切出来的数量并不一致。
输出侧按 max_tokens 上界估。输出长度事前不可知，保守做法是直接按你设的上限记账。这会高估，代价是限流器比实际更保守——但限流器保守是安全的方向，激进才危险。

估算不准怎么办？加一个校准回路：请求返回后拿到实际用量（多数平台会在响应里返回 usage 字段，具体字段名以各平台文档为准），把估算值和实际值的差补回桶里。多估了就还一部分令牌，少估了就多扣一点。跑一天下来，你的估算系数就有据可调了。

队列与背压：请求排不上的时候做什么

限流器只解决”什么时候放行”，它不回答”放不行的请求去哪了”。这个问题不回答清楚，前面所有工作都会在流量尖峰那天报废。

无界队列是隐患，不是缓冲

我见过最多的一种写法：所有请求丢进一个没有上限的队列，工作协程按限流器的节奏慢慢消费。看起来很优雅，实际上埋了三颗雷：

内存单调上涨。队列里存的不是几个字节，是完整的请求体——带着几千 token 的上下文、可能还有图片的 base64。尖峰持续几分钟，进程就 OOM 了。
延迟没有上限。队列越长，队尾的请求等得越久。当入队速率长期高于出队速率时，延迟是发散的，不是收敛到某个值。
在给已经走掉的用户干活。这是最浪费的一条。用户点了按钮，等了 8 秒没反应，关掉页面走了。而他的请求还老老实实排在队列第 400 位，10 分钟后轮到它，认认真真调用了上游，认认真真花了钱，然后把结果写进一个没人看的地方。

队列的作用是吸收短时尖峰，不是承载长期过载。 长期过载是容量问题，队列解决不了容量问题，它只能把容量问题伪装成延迟问题，然后在更糟的时机爆开。

该怎么做：有界 + 超时 + 明确状态

三件套，缺一不可。

import asyncio

class BoundedGate:
    def __init__(self, limiter: "DualLimiter", max_queue: int, max_wait_sec: float):
        self.limiter = limiter
        self.slots = asyncio.Semaphore(max_queue)   # 有界：排队位置也是有限资源
        self.max_wait = max_wait_sec

    async def submit(self, est_tokens: float, call):
        # 1) 抢不到排队位，直接快速失败，不要在这里等
        if not self.slots.locked() and self.slots._value > 0:
            await self.slots.acquire()
        else:
            raise QueueFull("排队位已满")
        try:
            # 2) 排队有超时上限，等不到就丢弃
            ok = await asyncio.to_thread(
                self.limiter.acquire, est_tokens, self.max_wait
            )
            if not ok:
                raise QueueTimeout(f"排队超过 {self.max_wait}s 未获得配额")
            return await call()
        finally:
            self.slots.release()

三个参数怎么定，各有各的依据：

参数	定它的依据
队列容量	单个排队请求占的内存 × 容量 ≤ 你愿意给出去的内存预算
排队超时	必须小于调用方的超时。调用方 10 秒超时，你排队 30 秒毫无意义
满了之后的行为	由业务决定：拒绝、降级到小模型、还是转异步

最后一行才是关键。“排不上怎么办”是业务问题，不是技术问题，限流器只能把它暴露出来让业务回答：

同步交互场景（用户在等）：宁可快速失败，给一个明确的”当前繁忙，请稍后重试”，也别让人干等。快速失败至少保住了用户的时间和你的一次调用费。
可降级场景：排不上大模型的队，就转去打小模型或者用缓存/规则兜底。这是最理想的一档，但要求业务侧真的能接受降级结果。
异步批处理场景（没人在等）：这才是队列真正该发挥作用的地方。转成任务、落库、慢慢消费，延迟上限可以放得很宽。

不管选哪个，要给上层返回一个能区分的状态。别把”被限流拒绝”和”上游报错”混成同一个异常类型往上抛——它们的处置方式完全不同，混在一起会让上层的重试逻辑做出错误的决定。

限流和重试的职责边界

限流在前，重试在后。这两个东西经常被写成一团，然后互相打架。

划清边界最简单的办法，是问一句”这次失败在时间轴上发生在哪”：

	限流器管的	重试管的
时机	请求发出之前	请求失败之后
依据	我预估的配额还够不够	上游实际返回了什么
手段	等、拒、降级	退避后重发
目标	尽量不产生 429	处理已经产生的 429 和 5xx

限流做对了，重试的触发次数会显著下降——因为大部分本来会撞 429 的请求，在客户端就被排住了。这时候重试才回归它本来的角色：处理那些你预测不到的情况（上游临时波动、5xx、网络抖动、你的配额估算偏差）。

两个常见的重复实现，看到就该合并：

重试里自己 sleep 一段固定时间来”避让”限流。这实际上是在写第二个限流器，而且是最差的那种（没有全局视野，每个失败的请求各睡各的，恢复的时候一起醒来，形成同步的重击）。退避该做的是抖动，不是限速。
限流器在被拒后自动重发。限流器不该发请求，它只该回答”能不能发”。让它自动重发，会导致调用方的超时预算被限流器悄悄吃掉，出问题时排查链路极其难受。

还有一件必须做的事：重试的请求要重新过限流器。这条经常被漏掉——重试逻辑拿着原来的请求直接重发，绕过了限流器，于是限流器的计数和实际打出去的量对不上，桶越算越乐观。退避、抖动、Retry-After 优先这些细节，重试与指数退避怎么写那篇有完整的写法。

用实际撞限率做反馈，动态调速

前面所有配置都建立在”我知道上游的配额是多少”这个前提上。现实里这个前提经常不成立：文档没写、账户等级变了、平台调整了策略、或者你跟别的团队共享同一个组织。

所以限流器最好别写死。把实际的撞限率当成反馈信号，让它自己找到边界。

最简单可行的策略是乘性减、加性增（AIMD），思路跟 TCP 拥塞控制一样：

撞到限了，速率乘以一个小于 1 的系数，比如砍到 70%。要砍得果断，因为你现在正在制造问题。
一段时间没撞限，速率加上一个固定小增量，比如每 30 秒加 5%。要加得克制，因为你不知道边界在哪，只能一点点试。

class AdaptiveRate:
    def __init__(self, init_rpm: float, floor: float, ceiling: float):
        self.rpm = init_rpm
        self.floor, self.ceiling = floor, ceiling
        self.last_up = time.monotonic()
        self._cooldown_until = 0.0

    def on_rejected(self):
        """撞限了：乘性减，并进入冷静期避免连续多次下调。"""
        now = time.monotonic()
        if now < self._cooldown_until:
            return
        self.rpm = max(self.floor, self.rpm * 0.7)
        self._cooldown_until = now + 10.0   # 冷静期，至少覆盖一次退避
        self.last_up = now

    def on_tick(self):
        """定期调用：加性增，且必须离上次下调足够远。"""
        now = time.monotonic()
        if now - self.last_up >= 30.0 and now >= self._cooldown_until:
            self.rpm = min(self.ceiling, self.rpm * 1.05)
            self.last_up = now

几个必须有的护栏：

下限（floor）。别让连续撞限把速率砍到接近零，那时候系统等于停摆，而且再也爬不回来。
上限（ceiling）。如果你知道官方配额，上限就设成配额的某个百分比，别一路往上加。
冷静期。一次尖峰会连续产生几十个 429，如果每个都触发一次乘性减，速率会在一秒内被砍到地板上。冷静期保证一轮事件只降一次。
变更要打日志。速率的每次调整都记一行，出事复盘的时候这是最有价值的线索——你能看出撞限是持续的还是脉冲的，是自己打太猛还是隔壁环境在抢。

反馈信号也不只有 429 一种。上游返回的用量字段、响应延迟的变化趋势，都可以纳进来。但先把最简单的那条跑起来：撞限就降，不撞就慢慢升。这一条能解决大部分问题，复杂策略往往是过度设计。

多副本部署下，单机限流会失效

这是最后一个坑，也是最多人栽的一个：在单进程里测得好好的限流器，一上 K8s 扩到 6 个副本，就全线失效了。

原因很朴素：每个副本各跑一份限流器，各自按完整配额限流。你配了 RPM 上限 300，副本数是 6，实际打出去的是 1800。而且这个问题在灰度阶段发现不了——1 个副本的时候一切正常，是扩容那一刻炸的。

两条路，各有代价。

路线一：按副本数分摊配额

每个副本只用总配额的 1/N。实现上几乎零成本：把副本数注入环境变量，启动时用 总配额 / 副本数 初始化限流器。

代价：

利用率低。流量在副本之间从来不是均匀的，某个副本闲着，它那份配额也浪费掉了。整体上你只能跑到总配额的一个不高的比例。
弹性伸缩时会算错。HPA 一扩容，如果副本数没同步到限流器，新老副本加起来又超了。得让副本数是动态读取的（比如通过 Downward API 或者服务发现），而不是启动时固定死。
副本挂掉时配额闲置。缩容或者 Pod 崩溃，那份配额没人接手。

适用：副本数稳定、流量相对均衡、你能接受牺牲一部分利用率换取零外部依赖。多数中小规模场景选这条就够了。

路线二：集中式计数

把令牌桶放到一个共享存储里（Redis 之类），所有副本对同一个桶做原子扣减。

# 用 Redis 脚本保证「补充 + 判断 + 扣减」的原子性（示意，非完整实现）
LUA = """
local key, rate, cap, now, need = KEYS[1], tonumber(ARGV[1]), tonumber(ARGV[2]),
                                  tonumber(ARGV[3]), tonumber(ARGV[4])
local t = redis.call('HMGET', key, 'tokens', 'ts')
local tokens = tonumber(t[1]) or cap
local ts = tonumber(t[2]) or now
tokens = math.min(cap, tokens + (now - ts) * rate)
if tokens >= need then
  tokens = tokens - need
  redis.call('HMSET', key, 'tokens', tokens, 'ts', now)
  return 1
end
redis.call('HMSET', key, 'tokens', tokens, 'ts', now)
return 0
"""

代价：

多一次网络往返。每个请求发出前先问一次 Redis。对于本身耗时几秒的大模型调用，这点开销可以忽略；对于高频小请求就要掂量了。
多一个故障点。Redis 挂了怎么办？必须想清楚降级行为：是全部放行（fail-open，风险是瞬间打爆上游），还是全部拒绝（fail-closed，风险是自己先跪）。我倾向 fail-open 但同时退回到本地的保守限流器——共享桶不可用时，每个副本按一个很小的本地配额继续跑，保证不会雪崩。
原子性必须做对。“读取-计算-写回”三步分开做，在并发下一定会超发。必须用脚本或者原生的原子命令把它压成一步。

适用：副本数多、流量倾斜严重、配额贵到不能浪费。代价是你要认真运维这个共享组件。

顺带一提：如果你已经在用统一网关来收敛所有模型调用，限流最合适的位置就是网关——它天然是集中的，不需要额外引入共享存储，而且能顺便把用量统计、故障转移一起做了。这条路的取舍在多供应商网关与故障转移里展开过。至于配额到底该按多少来规划，并发与成本怎么一起算那篇给了算账的口径。

不知道上游配额时，怎么稳妥起步

如果文档里查不到限速数值——这很常见，很多平台并不公开——也不用停在这里。按这个顺序走：

从很低的并发起步，比如同时在途 2 到 3 个请求。低到明显够不着任何配额。
先把观测建起来再加压。至少要能看到：请求速率、token 速率、429 计数、排队等待时长的分布。没有观测就加压，等于闭着眼睛开车。
每次只加一个维度，观察一段时间。撞到限了，那个位置就是边界的近似值，往回退一档记下来。
把探到的边界写成配置，而不是硬编码。平台会调整策略，你的账户等级也会变，这个数字迟早要改。
用小批量跑出自己的单位成本。在探边界的同时顺手把成本口径也标定了：一次典型请求平均多少 token、折算下来多少钱。这个数字比任何估算都可靠。

整个过程可能要花几天，但它换来的是一组属于你自己业务形态的真实参数，比抄来的配置有用得多。

自查清单

请求发出前有没有一道限流关卡？还是只在收到 429 之后才有动作？
限流是不是只做了 RPM 一个维度？token 维度（TPM）有没有单独的桶？
开发、CI、预发、生产是不是共用同一个组织/项目的配额？非生产流量有没有单独的、更小的客户端配额？
队列有没有容量上限？排队有没有超时？排队超时是不是小于调用方的超时？
排不上队的请求，返回给业务的状态能不能和”上游报错”区分开？
重试的请求有没有重新过限流器？重试里有没有偷偷写了第二套限速逻辑？
撞限率有没有接进反馈回路自动降速？降速有没有下限和冷静期，恢复是不是缓慢加回？
服务是多副本部署吗？限流器是按副本数分摊了配额，还是走集中式计数？扩容时这个数字会自动跟着变吗？

来源与边界：文中关于「速率限制适用于组织级别而非个别用户」「先撞到哪个阈值哪个生效」两条，出自 Groq 官方 rate limits 文档，仅作为限速模型的一个实例。各平台的限速维度、计数口径、生效层级（按 key / 按项目 / 按组织）都不相同，具体机制与数值一律以各平台官方文档和控制台为准。 本文未列出任何厂商的具体限速数值，文中代码为实现思路示意，需按你自己的运行环境调整。