# 三平台 API 限流怎么绕开瓶颈？三个参数调完就顺了（附 Python 源码）
先把话说清楚：**平台的限流绕不过去**。
改 IP、换 AppKey、注册多个应用分摊额度——这些做法短期有效，但只要平台做应用维度关联，一封就是全封，而且是永久性的。我们做这行两年，见过的封号案例里，一半是这么来的。
所以这篇文章讲的不是"绕开限制"，是**"别再浪费额度"**。
因为真实情况是：撞限流的团队里，**九成的瓶颈不是额度不够，是额度被烧在了没用的地方**。我们自己的数据：调参前一个月撞限流 3400 次，调参后 12 次——平台的额度一次没变，变的只是我们自己怎么用。
三个参数，加一套配额账本。下面是完整的东西。
---
## 一、先搞清楚你在跟什么限流打交道
三个平台的限流模型**长得完全不一样**。用一套逻辑套三个平台，是这个话题里最常见的错误。
| | 淘宝 | 京东 | 1688 |
| ------- | -------------- | ----------- | ---------- |
| 限流维度 | appkey × 接口 | 应用 × 接口 | appkey 流量包 |
| 形态 | 日调用量 + QPS 双限 | 日调用量 + QPS | 总量包 + QPS |
| 典型报错 | `code: 7`、`26` | 业务码 + `429` | `429`、业务码 |
| 失败调用算不算 | **算** | **算** | 部分算 |
| 恢复粒度 | 秒级 | 分钟级 | 秒级 |
| 是否可提额 | 可申请 | 走商务 | 可购买 |
这张表里有两行直接决定了架构，但大多数实现都踩了：
### 关键结论 1：限流是「接口 × 应用」维度的，不是全局的
淘宝的调用量配额是**按接口分别计**的。`item_get` 打满了，`trade_list` 一点不受影响。
所以全局令牌桶是错的。
我见过不止一个项目这么写：
```python
# ❌ 错的
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 关键词兜底：
>
> ```python
> 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 小时就只能干等着——而客户投诉通常发生在下午。
```python
# 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 × 平均耗时
```
也就是说：
```python
concurrency ≈ target_qps × p95_latency_seconds
```
举个具体的：目标 100 QPS，p95 延迟 300ms → `100 × 0.3 = 30`。
**为什么用 p95 不用平均值**：因为撞墙往往发生在长尾。平均 200ms、p95 800ms 的接口，你按平均值算出来 20，实际会遇到"20 个位置里 5 个卡了 8 秒"，然后队列瞬间堆起来，后面全在等。按 p95 算，留的余量正好覆盖长尾。
两个常见的错法：
| 错法 | 现象 | 后果 |
| ---------- | ---------- | ---------------------- |
| 并发给太大（200） | 90% 的请求在排队 | 内存和 fd 被占满，P99 恶化，还撞限流 |
| 并发给太小（10） | 打不满目标 QPS | 吞吐上不去，任务跑不完 |
第二种更容易被忽略。`concurrency=10`、延迟 300ms → 实际吞吐上限 33 QPS。你想跑 100 QPS，永远跑不到，而且看日志全是"正常返回"，没有报错。
```python
# 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)
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 倍。所以必须集中式。
```lua
-- 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}
```
```python
# 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（上面为了简洁省略了，生产环境要补）。
---
## 五、参数三：退避曲线——退多快，恢复多慢
前两个参数决定"会不会撞"，这一个决定"撞了之后多久恢复"。它是一对系数（降幅 + 恢复步长），合成一条曲线。
三种错法，按危害排序：
| 错法 | 代码 | 为什么错 |
| -------- | ------------------------ | ----------------------------------- |
| 固定 sleep | `await asyncio.sleep(1)` | 撞墙期间每秒都撞一次，1 秒变 1 小时；而且**每次失败都烧配额** |
| 直接放弃 | `raise` | 大批任务失败，客户重跑，额度烧得更多 |
| 无脑重试 | `for _ in range(5)` | 雪崩，还可能触发平台的"恶意调用"风控 |
正确做法是 AIMD（加法增、乘法减），但标准 AIMD 有个缺陷：**它不记得上次撞在哪**。
所以每次恢复都要重新"试探"过去，第二次撞在同一个位置。加一个临界点记忆就能解决：
```python
# 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` 之后，那个请求**不能再立刻重试**。它要退回去排队，等新速率生效。否则你降了速率，但手上这批已经在途的请求还在猛冲，等于没降。
---
## 六、把三个参数装到一起
```python
# 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
```
配置长这样（三个平台的值完全不一样，这就是为什么不能写死）：
```yaml
# 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 万订单同步），平台的额度一次没改：
| 指标 | 调参前 | 调参后 | 靠什么 |
| --------- | ------------- | --------------- | ------------------ |
| 撞限流次数 / 天 | 3,400 | **12** | burst 调小 + AIMD 记忆 |
| 平均 QPS | 38（峰值乱冲到 200） | **稳定 92** | 桶速率 + 并发按 p95 算 |
| 任务耗时 | 5h20m（一半在退避） | **1h05m** | 三件叠加 |
| 配额利用率 | 61%（失败也烧） | **97%** | 重试分级 + 分接口桶 |
| 配额燃烧比 | 下午 3 点就 > 1.5 | 全天稳定在 0.95~1.05 | burn_ratio 监控 |
最值得看的是第三行。**耗时从 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 曲线和平台报错分布，再定这三个数。*
