三平台 API 限流怎么绕开瓶颈？三个参数调完就顺了（附 Python 源码）

先把话说清楚：平台的限流绕不过去。

改 IP、换 AppKey、注册多个应用分摊额度——这些做法短期有效，但只要平台做应用维度关联，一封就是全封，而且是永久性的。我们做这行两年，见过的封号案例里，一半是这么来的。

所以这篇文章讲的不是"绕开限制"，是"别再浪费额度"。

因为真实情况是：撞限流的团队里，九成的瓶颈不是额度不够，是额度被烧在了没用的地方。我们自己的数据：调参前一个月撞限流 3400 次，调参后 12 次——平台的额度一次没变，变的只是我们自己怎么用。

三个参数，加一套配额账本。下面是完整的东西。

一、先搞清楚你在跟什么限流打交道

三个平台的限流模型长得完全不一样。用一套逻辑套三个平台，是这个话题里最常见的错误。

这张表里有两行直接决定了架构，但大多数实现都踩了：

关键结论 1：限流是「接口 × 应用」维度的，不是全局的

淘宝的调用量配额是按接口分别计的。item_get 打满了，trade_list 一点不受影响。

所以全局令牌桶是错的。

我见过不止一个项目这么写：

# ❌ 错的
limiter = TokenBucket(rate=50)   # 全局 50 QPS

async def call(platform, api, **kw):
    await limiter.acquire()
    ...

后果是：铺货任务在猛拉详情接口，把全局令牌吃光，这时候一个物流查询请求进来，明明平台侧完全不限它，却要排队等 3 秒。

正确做法：桶的 key 是 (platform, api)。粒度对齐平台的限流维度，不多不少。粒度再细（比如按 item_id）没意义，因为平台不按这个限；粒度再粗（按平台）就浪费了额度。

关键结论 2：失败的调用也占配额

这条最反直觉，也最贵。

淘宝的调用量是按调用次数算的，不看成功失败。所以你签名错了、参数漏了，重试三次——四次调用全部计入配额，全部是浪费。

这就是为什么上一篇讲的重试分级在这里格外重要：fatal 类错误一次重试都不该有。它不只是"没用"，它是在偷你还剩的额度。

顺带一个实现细节：不要只靠 code 判断错误类型。平台会新增错误码，你的映射表永远追不上。要加一层 message 关键词兜底：

def is_rate_limited(code: str, msg: str) -> bool:
    if code in ("7", "26", "429"):
        return True
    kw = ("limit", "too many", "频繁", "超限", "exceed", "throttl")
    return any(k in msg.lower() for k in kw)

二、瓶颈在哪：先算三笔账，再动参数

不先算账就调参，等于闭着眼睛拧螺丝。这三笔账都是 SQL 或日志就能算出来的，半小时的事。

账 1：配额花在哪个接口上了

接口              调用次数    占比     带来有效数据
item_get         412,000    68.3%    41,200 条
trade_list        88,000    14.6%    88,000 条
logistics_trace   71,000    11.8%     9,400 条   ← 注意这里
shop_info         32,000     5.3%    32,000 条

看到问题了吗：logistics_trace 占了 11.8% 的配额，只换来 9,400 条有效数据。因为大部分查询打在了已签收的订单上——早就没有新轨迹了。

（这就是第三篇讲的：用 INSERT OR IGNORE + rowcount == 1 去重，加上 last_status != 'SIGNED' 过滤，这个接口的调用量能降 80%。降调用量比调参数有效一百倍，因为调参只是在分配额度，降量是在释放额度。）

账 2：有效请求率

有效请求率 = 真正拿到新数据的调用 / 总调用

低于 30% 说明你在"重复问同样的问题"，该加缓存或做增量了。这个数字跟限流无关，但它决定了你需要多少额度。

账 3：配额燃烧曲线

这是最容易被忽略的一笔。日配额 10 万，如果前 6 小时用掉 8 万，剩下 18 小时就只能干等着——而客户投诉通常发生在下午。

# throttle/burn.py
from datetime import datetime
from zoneinfo import ZoneInfo

TZ = ZoneInfo("Asia/Shanghai")

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def consume_quota(r, platform: str, api: str, n: int = 1) -> None:
    """每次调用都记账，按自然日归零"""
    day = datetime.now(TZ).strftime("%Y%m%d")
    key = f"quota:{platform}:{api}:{day}"
    await r.incrby(key, n)
    await r.expire(key, 90000)       # 25 小时，跨天自动过期


async def burn_ratio(r, platform: str, api: str, daily_quota: int) -> float:
    """
    燃烧比 = 已用比例 / 时间进度比例
    > 1.3 报警：按这个速度，今天不够用
    < 0.5 提示：额度有富余，可以提高并发
    """
    day = datetime.now(TZ).strftime("%Y%m%d")
    used = int(await r.get(f"quota:{platform}:{api}:{day}") or 0)

    now = datetime.now(TZ)
    elapsed = (now.hour * 3600 + now.minute * 60 + now.second) / 86400
    progress = max(elapsed, 1 / 1440)          # 防止零点除零

    return (used / daily_quota) / progress

这个 burn_ratio 是我们监控里最有用的一个指标。它能提前 4 小时预测"今天配额会不够"，而不是等到撞墙了才知道。

而且它是双向的：比值 0.5 说明额度有富余，那段时间可以把并发调高一点，把活干完。

三、参数一：concurrency——它不是一个可以随便拍的数

标题说三个参数，第一个就打破你的直觉：并发不是自变量，它是因变量。

恒等式：

在途请求数 = QPS × 平均耗时

也就是说：

concurrency ≈ target_qps × p95_latency_seconds

举个具体的：目标 100 QPS，p95 延迟 300ms → 100 × 0.3 = 30。

为什么用 p95 不用平均值：因为撞墙往往发生在长尾。平均 200ms、p95 800ms 的接口，你按平均值算出来 20，实际会遇到"20 个位置里 5 个卡了 8 秒"，然后队列瞬间堆起来，后面全在等。按 p95 算，留的余量正好覆盖长尾。

两个常见的错法：

第二种更容易被忽略。concurrency=10、延迟 300ms → 实际吞吐上限 33 QPS。你想跑 100 QPS，永远跑不到，而且看日志全是"正常返回"，没有报错。

# throttle/gate.py
import asyncio


class GateBusy(Exception):
    pass


class Gate:
    """
    每个 (platform, api) 一个 Gate。
    并发 + 令牌桶 + AIMD，三件东西绑在一起，因为它们必须同步调整。
    """

    def __init__(self, name: str, bucket, adaptive, concurrency: int):
        self.name = name
        self.bucket = bucket
        self.adaptive = adaptive
        self.concurrency = concurrency
        self._sem = asyncio.Semaphore(concurrency)
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    async def __aenter__(self):
        # 先占并发槽，再等令牌。
        # 反过来的话，等令牌的请求会把槽全占住，
        # 明明桶里还有令牌也没人能拿到（真实的坑，别问怎么知道的）
        try:
            await asyncio.wait_for(self._sem.acquire(), timeout=1.0)
        except asyncio.TimeoutError:
            raise GateBusy(f"{self.name} concurrency={self.concurrency} 已满")

        try:
            await self.bucket.wait(1)
        except Exception:
            self._sem.release()
            raise
        return self

    async def __aexit__(self, exc_type, exc, tb):
        self._sem.release()
        if exc is None:
            await self.adaptive.on_success()
        elif getattr(exc, "rate_limited", False):
            await self.adaptive.on_limited()
        return False

注意 __aenter__ 里那个顺序，以及那个 1 秒超时。并发满了就立刻告诉调用方"忙"，别让它排队——排队会掩盖真实压力，等服务崩了你才发现队列已经三千长了。

四、参数二：令牌桶——rate 管长期，burst 管瞬时

这两个是文章标题里的第二个参数，但其实是两个数：速率决定长期，桶深决定瞬时。

为什么必须有 burst

电商接口调用是脉冲式的，不是均匀分布的。一次铺货 500 个商品，就是突然一批并发请求涌进来。你的令牌桶如果 burst = rate（每秒回一个就只能存一个），那这 500 个请求会被均匀摊到 500/rate 秒里——本来 10 秒能干完的活，变成 50 秒。

burst 该多大

两个约束同时要满足：

① burst ≥ 平台单秒允许的突发量 × 0.5   （别浪费突发能力）
② burst + rate × 1s ≤ 平台单秒硬上限    （别打穿）

经验值 burst = rate × 2 ~ 5。太小失去突发能力，太大直接撞墙。

举个反例：我们最早给淘宝 item_get 设 rate=80, burst=500。因为那个接口日配额够大，想着"一次多放点没关系"。结果每 6 秒就有一次 500 连发的脉冲，平台侧按秒窗口算，直接 429。rate 明明只有 80，但瞬时到了 500。

分布式令牌桶（Lua + Redis）

单机令牌桶在多实例部署下是灾难：4 个 worker 各算各的，实际 QPS 是配置值的 4 倍。所以必须集中式。

-- throttle/bucket.lua
-- KEYS[1] 桶 key
-- ARGV[1] rate（每秒回多少）ARGV[2] burst（最大存量）
-- ARGV[3] now_ms             ARGV[4] want（这次要几个）
local rate  = tonumber(ARGV[1])
local burst = tonumber(ARGV[2])
local now   = tonumber(ARGV[3])
local want  = tonumber(ARGV[4])

local d      = redis.call('HMGET', KEYS[1], 'tokens', 'ts')
local tokens = tonumber(d[1])
local ts     = tonumber(d[2])

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

local delta = math.max(0, now - ts) / 1000
tokens = math.min(burst, tokens + delta * rate)

if tokens < want then
  local need_ms = math.ceil((want - tokens) / rate * 1000)
  return {
   0, need_ms}
end

tokens = tokens - want
redis.call('HMSET', KEYS[1], 'tokens', tokens, 'ts', now)
redis.call('PEXPIRE', KEYS[1], math.ceil(burst / rate * 1000) + 5000)
return {
   1, 0}

# throttle/bucket.py
import asyncio
import time
from pathlib import Path


class RateLimitTimeout(Exception):
    def __init__(self, key: str, wait: float):
        super().__init__(f"{key} 排队超时，还需 {wait:.2f}s")
        self.rate_limited = True


class TokenBucket:
    def __init__(self, redis, key: str, rate: float, burst: int):
        self.r = redis
        self.key = f"rl:{key}"
        self.rate = rate
        self.burst = burst
        self._lua = Path("throttle/bucket.lua").read_text(encoding="utf-8")
        self._sha_id: str | None = None

    async def _script_sha(self) -> str:
        if self._sha_id is None:
            self._sha_id = await self.r.script_load(self._lua)
        return self._sha_id

    async def acquire(self, want: int = 1) -> float:
        """返回 0 表示拿到；> 0 表示还差多少秒"""
        now_ms = int(time.time() * 1000)
        ok, wait_ms = await self.r.evalsha(
            await self._script_sha(), 1, self.key,
            self.rate, self.burst, now_ms, want,
        )
        return 0.0 if ok else wait_ms / 1000

    async def wait(self, want: int = 1, timeout: float = 60.0) -> None:
        deadline = time.monotonic() + timeout
        while True:
            delay = await self.acquire(want)
            if delay == 0:
                return
            if time.monotonic() + delay > deadline:
                raise RateLimitTimeout(self.key, delay)
            await asyncio.sleep(delay)

一个容易忽略的点：lua 脚本用 evalsha 而不是 eval。eval 每次都传脚本全文，一个大脚本几百字节，高频调用时这部分网络开销不小；evalsha 只传 40 字节的 sha1。用 script_load 缓存，节点重启后 evalsha 会报 NOSCRIPT，需要捕获后重新 load（上面为了简洁省略了，生产环境要补）。

五、参数三：退避曲线——退多快，恢复多慢

前两个参数决定"会不会撞"，这一个决定"撞了之后多久恢复"。它是一对系数（降幅 + 恢复步长），合成一条曲线。

三种错法，按危害排序：

正确做法是 AIMD（加法增、乘法减），但标准 AIMD 有个缺陷：它不记得上次撞在哪。

所以每次恢复都要重新"试探"过去，第二次撞在同一个位置。加一个临界点记忆就能解决：

# throttle/adaptive.py
import time


class AdaptiveRate:
    """
    AIMD + 临界点记忆

    撞限流     → rate *= 0.5        （乘法降，退得快）
    连续成功 N → rate += step        （加法升，进得慢）
    记忆       → 记下撞墙时的 rate，恢复不超过它的 90%

    最后这条是关键：没有它，每次恢复都会重新撞在同一个地方。
    """

    def __init__(self, rate: float, min_rate: float = 1.0,
                 max_rate: float = 500.0, recover_after: int = 50,
                 step: float = 1.0, memory_ratio: float = 0.9):
        self.rate = rate
        self.min_rate = min_rate
        self.max_rate = max_rate
        self.recover_after = recover_after
        self.step = step
        self.memory_ratio = memory_ratio

        self.ceiling: float | None = None    # 上次撞墙时的速率
        self._ok_streak = 0
        self._last_hit = 0.0

    async def on_success(self):
        self._ok_streak += 1
        if self._ok_streak < self.recover_after:
            return
        self._ok_streak = 0

        new_rate = self.rate + self.step
        if self.ceiling is not None:
            new_rate = min(new_rate, self.ceiling * self.memory_ratio)
        self.rate = min(self.max_rate, new_rate)

    async def on_limited(self):
        now = time.monotonic()
        # 5 秒内的第二次撞击才更新 ceiling，
        # 避免一次偶发毛刺把 ceiling 永久压低
        if now - self._last_hit > 5.0 or self.ceiling is None:
            self.ceiling = max(self.ceiling or 0, self.rate)
        self._last_hit = now

        self._ok_streak = 0
        self.rate = max(self.min_rate, self.rate * 0.5)

memory_ratio=0.9 这个值有讲究：太低（0.5）恢复太保守，明明还有余量却上不去；太高（0.99）会频繁在边界试探。0.9 是实测下来最平衡的——撞墙一次后，稳定在临界值的 90%，之后很少再撞。

还有一个必须做但很少有文章提的点：on_limited 之后，那个请求不能再立刻重试。它要退回去排队，等新速率生效。否则你降了速率，但手上这批已经在途的请求还在猛冲，等于没降。

六、把三个参数装到一起

# throttle/registry.py
import asyncio


class Throttle:
    """
    按 (platform, api) 生成 Gate。
    参数不是写死的常量，而是从配置里读——因为三个平台的三个参数值完全不同。
    """

    def __init__(self, redis, config: dict):
        self.r = redis
        self.cfg = config
        self._gates: dict[tuple[str, str], Gate] = {
   }
        self._lock = asyncio.Lock()

    async def get(self, platform: str, api: str) -> Gate:
        key = (platform, api)
        gate = self._gates.get(key)
        if gate is not None:
            return gate

        async with self._lock:                       # 双检锁，避免并发建重复桶
            gate = self._gates.get(key)
            if gate is not None:
                return gate

            c = self.cfg.get(platform, {
   }).get(api, self.cfg["default"])
            bucket = TokenBucket(self.r, f"{platform}:{api}",
                                 rate=c["rate"], burst=c["burst"])
            gate = Gate(
                name=f"{platform}/{api}",
                bucket=bucket,
                adaptive=AdaptiveRate(rate=c["rate"]),
                concurrency=c["concurrency"],
            )
            self._gates[key] = gate
            return gate

    async def call(self, platform: str, api: str, fn, **kwargs):
        gate = await self.get(platform, api)

        # 速率被 AIMD 调过之后，令牌桶要跟上
        gate.bucket.rate = gate.adaptive.rate

        async with gate:
            try:
                return await fn(**kwargs)
            except Exception as e:
                if getattr(e, "rate_limited", False):
                    # 这里退避一下，让降速生效再回来
                    await asyncio.sleep(1.0)
                raise

配置长这样（三个平台的值完全不一样，这就是为什么不能写死）：

# throttle.yaml
default:
  rate: 20
  burst: 60
  concurrency: 8

taobao:
  item_get:      {
    rate: 80, burst: 300, concurrency: 24 }   # 日配额大，可冲
  trade_list:    {
    rate: 30, burst: 90,  concurrency: 9 }    # 敏感接口，保守
  logistics_trace: {
    rate: 10, burst: 20, concurrency: 3 }   # 限流最严

jd:
  item_get:      {
    rate: 40, burst: 120, concurrency: 12 }
  order_list:    {
    rate: 15, burst: 45,  concurrency: 5 }

alibaba:
  item_get:      {
    rate: 25, burst: 75,  concurrency: 8 }

七、调完之后的数字

同一批任务（10 万商品铺货 + 3 万订单同步），平台的额度一次没改：

最值得看的是第三行。耗时从 5 小时 20 分降到 1 小时 05 分，但 QPS 只从 38 涨到 92——按比例算应该只快 2.4 倍，实际快了 4.9 倍。

多出来的那部分是"不再退避"。调参前那 5 小时里，有一半时间是在 sleep 等重试。限流最大的成本不是吞吐低，是时间被 sleep 吃掉了。

八、四个反直觉的结论

1. 并发调大不会更快，只会让长尾更慢。
并发超过 QPS × p95延迟 之后，多出来的请求只是在队列里等着，同时占用内存和文件句柄。真正的吞吐由令牌桶的 rate 决定，不由并发决定。

2. burst 是三个参数里最危险的一个。
rate 决定长期，出不了大事；burst 决定瞬时，设大了直接打穿平台窗口。而且它出问题的表现很迷惑——平均 QPS 明明没超，但就是天天撞 429。

3. 撞限流后立刻降速，恢复反而更快。
因为撞墙期间每次失败都在烧配额，而且失败请求也要等超时。降速看着"退了一步"，实际上是止损。

4. 减少请求数比调任何参数都有效。
调参最多把效率从 61% 提到 97%，这是 1.6 倍。而把物流查询从 71,000 次降到 12,000 次，是 6 倍的额度释放。调参是分配问题，降量是供给问题。 先做降量。

九、最后

三个参数，说到底是在回答三个问题：

• concurrency：同时在途多少个（由 QPS 和延迟推出来，不是拍的）
• rate / burst：每秒放多少个，能突然放多少个（长期与瞬时的分离）
• backoff：撞了之后退多快、恢复多慢（决定故障时长，不是决定吞吐）

再加一个不在标题里但最重要的：先算账。三笔账算完，你大概会发现真正的瓶颈不在限流上。

最后再说一次开头那句话：这套东西的目标是把额度用满，不是突破额度。前者是工程，后者是违规。我们做接口服务这些年，见过太多为了省点调用量去搞多 AppKey 分摊的，最后账号全封、客户数据全断——那个损失比省下来的钱大得多。

代码里 bucket.lua 和 adaptive.py 是不依赖业务的，可以单独抽出来用；registry.py 的配置结构按你们自己的接口清单改。所有阈值都建议先跑一周观察日志再定，别照抄我这几个数——平台额度、你的客户结构、调用模式都不一样。

附：文中淘宝 item_get 的 rate=80 是我们账号的实际额度推导出来的，不同账号、不同应用等级差别很大。接入前先用第七篇那套 StageTimer 跑一天，看真实 QPS 曲线和平台报错分布，再定这三个数。