# 三平台 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 曲线和平台报错分布，再定这三个数。*

# 接口稳定率 99.9% 背后，我们在链路层做了这 5 件事（附 Python 源码）

先说个数字：99.9% 的可用性，摊到 30 天里是 **43 分钟**。

43 分钟，一个月。听起来还有余量对吧？但这是**全部**——三个平台加在一起，所有接口、所有客户、所有时段，一个月累计只能有 43 分钟不达标。

我们做电商数据接口（淘宝 / 1688 / 京东），上游是别人家的系统。上游什么时候抖动、什么时候限流、什么时候半夜改字段，你控制不了。所以这个数字不可能靠"不出事"达成，只能靠**出事的时候影响面被限制住**。

这篇文章讲的就是"限制影响面"这件事：在业务代码和平台 API 之间加的那一层，我们叫**链路层**，里面做了五件事。

代码是 Python + asyncio + httpx，可以直接拿去用。

---

## 一、先把 99.9% 的口径定下来，否则没法谈

这是最多人跳过的一步，也是后面所有优化没意义的原因。

同一个系统，三种算法能算出三个数：

| 口径        | 怎么算               | 数字好看程度 | 有没有意义              |
| --------- | ----------------- | ------ | ------------------ |
| 进程存活率     | 服务没挂就算成功          | 99.99% | 基本没意义，挂了才叫故障       |
| HTTP 成功率  | 返回 200 就算成功       | 99.5%  | 有点意义，但 200 里可能有错误体 |
| **业务成功率** | 拿到**正确的业务数据**才算成功 | 99.9%  | 有意义，但最难做           |

我们说的是第三种。定义写清楚：

```
可用性 = 成功请求数 / 应当成功的请求数
```

两个关键词要解释：

**"成功"** —— 不是 HTTP 200，是拿到结构完整、字段非空、语义正确的业务数据。淘宝的接口 HTTP 永远 200，错误信息藏在 body 的 `error_response` 里。只统计状态码，你的监控永远是绿的。

**"应当成功"** —— 这个限定词救了我们。客户传错参数、AppKey 拼错、没授权就调接口，这些请求怎么算都是失败，而且**再怎么优化链路也不会成功**。

真实情况是：这类"调用方错误"占我们总请求量的 **6% 左右**。如果把它们算进分母，可用性的理论上限就是 94%，永远上不了 99.9%。

所以口径是：调用方错误**排除出分母**，但**单独监控**。一旦这个比例突然从 6% 涨到 15%，说明文档有问题，或者某个客户在乱调，得去查（通常是我们自己的锅：接口文档写得不清楚）。

> 这一步的价值不在于算出什么数字，而在于**逼你想清楚"失败"到底指什么**。想清楚了，后面五件事才知道该往哪使劲。

---

## 二、链路层在哪，它管什么

```
业务代码（下单、铺货、同步）
        │  它只知道：调 get_item(id)，要么拿到数据，要么拿到异常
        ▼
┌──────────────────────────────────────────┐
│  链路层（本文主角）                        │
│  ① 舱壁隔离  ② 超时预算  ③ 重试分级        │
│  ④ 熔断降级  ⑤ 探测与归一                  │
└──────────────────────────────────────────┘
        │  三平台各自 client，各自限流，各自熔断
        ▼
  淘宝开放平台 / 京东宙斯 / 1688 网关
```

三条设计原则，后面每个模块都在守：

1. **业务代码看不见重试、限流、超时**——它只描述"我要什么"，不描述"失败了怎么办"。
2. **故障隔离在平台粒度**——淘宝挂了不能拖累京东，京东限流不能让淘宝的请求排队。
3. **每一层都必须能"快速失败"**——宁可 100ms 返回一个明确的"现在不行"，也不要 30s 后返回一个超时。两者对业务的意义完全不同。

---

## 三、第 1 件事：舱壁隔离，让一个平台的病不传染

最经典的翻车场景：三个平台共用一个 `httpx.AsyncClient`，淘宝那侧开始超时，连接全部卡在淘宝上，连接池占满，京东的请求连连接都拿不到——**京东明明好好的，也跟着挂**。

解法是造船上的"舱壁"：每个平台一套独立的资源，进水只淹一个舱。

```python
# link/bulkhead.py
import asyncio
import httpx


class BulkheadFull(Exception):
    def __init__(self, name: str, concurrency: int):
        super().__init__(f"bulkhead {name} full (concurrency={concurrency})")

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class Bulkhead:
    """舱壁：每个平台一套并发额度，互不挤占"""

    def __init__(self, name: str, concurrency: int = 16):
        self.name = name
        self.concurrency = concurrency
        self._sem = asyncio.Semaphore(concurrency)

    async def __aenter__(self):
        try:
            # 注意这里是 0.5s，不是无限等
            # 排队本质上也是一种失败，只是它看起来像"在正常工作"
            await asyncio.wait_for(self._sem.acquire(), timeout=0.5)
        except asyncio.TimeoutError:
            raise BulkheadFull(self.name, self.concurrency)
        return self

    async def __aexit__(self, *exc):
        self._sem.release()
        return False


def make_client(platform: str) -> httpx.AsyncClient:
    """每个平台一个 client，参数不是随便填的"""
    return httpx.AsyncClient(
        http2=True,
        limits=httpx.Limits(
            max_connections=32,
            max_keepalive_connections=16,
            # ↓ 这一行是本篇最值钱的一行，下面有血案
            keepalive_expiry=25.0,
        ),
        timeout=httpx.Timeout(connect=2.0, read=8.0, write=2.0, pool=1.0),
        follow_redirects=False,
    )
```

三个参数值得单独说：

**`concurrency` 怎么定**：`QPS × 平均耗时`。假设某平台限流 100 QPS，平均响应 300ms，那理论并发是 30。取 16 是留了余量——并发给多了不会更快，因为**限流才是物理下限**，多出来的只是在网关排队，还白占内存。

**`pool` 超时 1 秒**：这是"拿不到连接"的等待上限。默认不设或者设很长，意味着连接池被占满时新请求会一直等，等到天荒地老，然后一起超时。1 秒拿不到就直接认输，快得多。

**`keepalive_expiry=25`**：这一行有故事，我单独讲。

### 血案复盘：一次因为 35 秒导致的成功率暴跌

某个凌晨，京东侧的成功率从 99.95% 掉到 92%，持续了两个小时。监控上没有任何"错误"，延迟 P99 从 400ms 涨到 8s。

查出来的原因很蠢：

- httpx 默认 `keepalive_expiry=5.0`，我们当时手写成了 `60.0`，觉得"连接复用久一点更好"
- 京东网关侧的空闲连接回收是 **30 秒**
- 于是 30~60 秒这个区间里的连接，是**对端已经关掉、我们以为还活着**的死连接
- 请求发出去，一直等 `read` 超时（8 秒），然后重试，重试又可能拿到另一个死连接

一个死连接 = 8 秒浪费。那晚 60% 的请求撞上了。

修法就是 `keepalive_expiry` 必须**小于**对端的空闲回收时间。拿不到对端确切值时，取 20~30 秒是最安全的区间。另外加一条：`RemoteProtocolError` 必须**不消耗重试次数**——那是连接级别的偶发问题，换条连接就好，不该算成"这次调用失败了一次"。

> 这个坑的可怕之处：**它不报错，只是慢**。你的错误率监控完全正常，只有延迟曲线在涨。如果你只看错误率不看延迟分布，这类问题能藏几个月。

---

## 四、第 2 件事：超时三级拆分 + 预算传递

"超时设 30 秒"是另一个常见错误。它有个隐藏假设：**一次业务调用只有一个请求**。

实际上一次铺货可能串 20 个请求。每个都 30 秒，最坏情况 10 分钟——用户早就关了页面，你的线程还在跑。

两个动作：

**（1）超时按接口类型拆开，不是一刀切**

| 接口类型      | connect | read | 理由             |
| --------- | ------- | ---- | -------------- |
| 列表 / 增量查询 | 2s      | 3s   | 数据量小，慢了就是有问题   |
| 商品详情      | 2s      | 6s   | 详情字段多，正常要 1~2s |
| 批量提交 / 导出 | 2s      | 15s  | 平台侧真的在跑任务      |
| 物流轨迹      | 2s      | 4s   | 单条查询，快         |

**（2）总预算往下传，剩余不够就立刻放弃**

```python
# link/budget.py
import time


class Budget:
    """
    一次业务调用分给链路的剩余时间。
    核心思想：超时不是「每个请求各自计时」，是「一共就这么久，你自己分」。
    """

    __slots__ = ("_deadline", "_total")

    def __init__(self, total_ms: int):
        self._total = total_ms / 1000
        self._deadline = time.monotonic() + self._total

    def left(self) -> float:
        return self._deadline - time.monotonic()

    def slice(self, want: float) -> float:
        """子请求可用超时：不超过剩余预算，也不低于 200ms"""
        return max(0.2, min(want, self.left()))

    def enough(self, need: float) -> bool:
        """还剩的时间够不够再试一次？不够就别试了"""
        return self.left() >= need
```

用起来是这样：

```python
budget = Budget(total_ms=8000)          # 这个业务动作总共 8 秒
for item_id in item_ids:
    if not budget.enough(1.0):          # 连 1 秒都不剩了
        raise BudgetExhausted(f"剩 {budget.left():.2f}s，还有 {len(rest)} 个没处理")
    await gw.call("item_get", timeout=budget.slice(6.0), ...)
```

**为什么要这么麻烦**：因为"超时"和"重试"必须联动。预算只剩 1 秒，你还按 6 秒超时去发请求、失败了再重试三次——这 1 秒变成了 20 秒，而且必然失败。有了预算，链路层就能回答一个业务层答不出的问题：**现在还有没有必要试**。

> 一个实现细节：`asyncio.wait_for` 超时之后，底层请求**不一定**立刻取消干净（尤其是 HTTP/2 多路复用下）。所以 httpx 自己的 `timeout` 必须比 `wait_for` 小一点，让 httpx 先超时、自己清理连接；`wait_for` 只做最后兜底。反过来用，会漏连接。

---

## 五、第 3 件事：重试分级——别把所有错误都当"再试一次"

先说最反直觉的一点：**大部分重试是无效的，而且有害**。

签名错了，重试一百次还是签名错，只是把配额烧光。参数少了，重试一百次还是少。权限没开，重试一百次还是没开。这些错误重试的唯一效果是：**把一次明确的失败，变成一次很慢的失败**。

所以第一件事是把三平台几百个错误码压成几类：

```python
# link/errors.py
from enum import Enum
import httpx


class Kind(str, Enum):
    OK = "ok"
    RETRY = "retry"      # 瞬时故障：网络抖动、5xx、网关超时 → 可立刻重试
    SLOW = "slow"        # 限流、配额耗尽 → 退避久一点再试
    FATAL = "fatal"      # 参数 / 签名 / 权限 → 重试无意义，直接抛
    UNKNOWN = "unknown"  # 没见过 → 不重试，立即告警，人工看
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex

# ---- 平台业务码 → 归一 ----
_TAOBAO = {
    "7":  Kind.SLOW,    # app call limited（限流）
    "15": Kind.RETRY,   # remote service error
    "21": Kind.FATAL,   # missing params
    "22": Kind.FATAL,   # invalid params
    "26": Kind.SLOW,    # 调用量超限
    "27": Kind.FATAL,   # 权限不足
    "28": Kind.FATAL,   # 应用被禁用
    "33": Kind.FATAL,   # 签名错误
    "40": Kind.FATAL,   # 会话过期
}

_JD = {
    "1001": Kind.FATAL,
    "1002": Kind.FATAL,
    "2001": Kind.RETRY,
    "3001": Kind.SLOW,
}

_ALIBABA = {
    "400": Kind.FATAL,
    "401": Kind.FATAL,
    "429": Kind.SLOW,
    "500": Kind.RETRY,
}

_BY_PLATFORM = {"taobao": _TAOBAO, "jd": _JD, "alibaba": _ALIBABA}


def classify(platform: str, exc: Exception | None = None,
             code: str | None = None) -> Kind:
    if exc is not None:
        if isinstance(exc, (httpx.ConnectTimeout, httpx.ReadTimeout,
                            httpx.WriteTimeout, httpx.PoolTimeout,
                            httpx.ConnectError)):
            return Kind.RETRY
        # 连接被对端关掉：换条连接就好，不是业务失败
        if isinstance(exc, httpx.RemoteProtocolError):
            return Kind.RETRY
        if isinstance(exc, httpx.HTTPStatusError):
            s = exc.response.status_code
            if s == 429:
                return Kind.SLOW
            if 500 <= s < 600:
                return Kind.RETRY
            return Kind.FATAL
        return Kind.UNKNOWN
    return _BY_PLATFORM.get(platform, {}).get(str(code), Kind.UNKNOWN)
```

第二件事是给重试加**全局预算**。这一条很多人没做，但它是防止"重试风暴"的唯一手段。

原理解释一下：上游开始抖动 → 10% 请求失败 → 重试 → 变成 30% 请求量 → 上游更扛不住 → 失败率涨到 30% → 重试 → 请求量翻倍……**重试本身成了压垮上游的那根稻草**。这就是雪崩。

```python
# link/retry.py
import random
import time


class RetryBudget:
    """
    全局重试预算：按速率发令牌，用完就不许重试。
    正常时几乎无感；上游抖动时它会自动把重试量摁住。
    """

    def __init__(self, capacity: int = 20, window: float = 10.0):
        self.capacity = capacity          # 窗口内最多允许多少次重试
        self.window = window
        self._tokens = float(capacity)
        self._last = time.monotonic()

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

    def allow(self) -> bool:
        self._refill()
        if self._tokens < 1:
            return False
        self._tokens -= 1
        return True


def backoff(attempt: int, base: float = 0.2, cap: float = 4.0) -> float:
    """
    指数退避 + full jitter
    full jitter 很重要：不加随机的话，一批同时失败的请求会同时重试，
    形成周期性的尖峰，把上游二次打挂
    """
    return random.uniform(0, min(cap, base * 2 ** attempt))
```

第三件事，也是最容易被忽略的：**写接口不能盲目重试**。

`item_update` 这种接口，超时了不代表没执行成功。可能平台已经改完价了，只是响应没回来。你重试一次，就改了两次——如果平台侧按"每次调用扣一次额度"计费，或者更新操作会触发审核流程，这就是实打实的损失。

规则：

- **读接口**：可以重试，最多 3 次
- **写接口**：必须带业务幂等键（`out_request_id`，用你自己的订单号/任务号），并且只在**确认没送达**的情况下重试（`ConnectError` 可以，`ReadTimeout` 不行）
- **没有幂等键的写接口**：一次都不重试，失败直接抛给业务层，让业务层用业务状态去判断

> 还有一个结构性问题：**重试必须收口在一层**。如果适配器里重试 3 次、网关层再重试 3 次、业务层还重试 3 次，一次调用放大成 27 次。这个问题在代码 review 时看不出来，只有在大促当天能看出来——那时候你已经没时间改了。

---

## 六、第 4 件事：熔断，加上"带标记的降级"

重试解决"偶发失败"，熔断解决"持续失败"。

区别在于：上游只是抖一下，重试有用；上游已经挂了，重试只是在浪费自己的时间。熔断器就是那个替你判断"它是不是已经挂了"的东西。

```python
# link/breaker.py
import time
from collections import deque


class Breaker:
    """
    closed    → 正常放行，滑窗内失败率超阈值 → open
    open      → 直接快速失败（不请求上游），冷却 30s → half_open
    half_open → 只放 3 个探测，2 个成功 → closed，任一失败 → open

    注意 half_open 放 3 个而不是 1 个：
    只放 1 个的话，那个请求恰好赶上抖动，就要再等一个完整冷却期，
    恢复会变得非常慢。这是实测出来的参数，不是拍脑袋。
    """

    def __init__(self, window=10.0, min_samples=20, fail_ratio=0.5,
                 cooldown=30.0, probe_quota=3, probe_need=2):
        self.window = window
        self.min_samples = min_samples        # 样本太少不算，避免启动就被熔断
        self.fail_ratio = fail_ratio
        self.cooldown = cooldown
        self.probe_quota = probe_quota
        self.probe_need = probe_need

        self.state = "closed"
        self._events: deque[tuple[float, bool]] = deque()
        self._opened_at = 0.0
        self._probes_left = 0
        self._probe_ok = 0

    def allow(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            if time.monotonic() - self._opened_at >= self.cooldown:
                self.state = "half_open"
                self._probes_left = self.probe_quota
                self._probe_ok = 0
                return True
            return False
        if self._probes_left > 0:
            self._probes_left -= 1
            return True
        return False

    def record(self, ok: bool):
        now = time.monotonic()

        if self.state == "half_open":
            if ok:
                self._probe_ok += 1
                if self._probe_ok >= self.probe_need:
                    self.state = "closed"
                    self._events.clear()
            else:
                self._trip(now)
            return

        self._events.append((now, ok))
        cutoff = now - self.window
        while self._events and self._events[0][0] < cutoff:
            self._events.popleft()

        if len(self._events) >= self.min_samples:
            fails = sum(1 for _, ok_ in self._events if not ok_)
            if fails / len(self._events) > self.fail_ratio:
                self._trip(now)

    def _trip(self, now: float):
        self.state = "open"
        self._opened_at = now
        self._events.clear()
```


然后是全篇我最想讲的部分：**降级不是返回错误，是返回"能用的旧数据 + 一个标记"**。

```python
# link/degrade.py
import json
import time
from dataclasses import dataclass


@dataclass
class Stale:
    value: dict
    age: float          # 数据多少秒前抓的
    reason: str

    def as_payload(self) -> dict:
        return {
            "data": self.value,
            "stale": True,
            "age": round(self.age, 1),
            "reason": self.reason,
        }


class StaleCache:
    """
    「最后一份好数据」，和业务缓存分开存。
    区别：业务缓存可以被主动失效（因为有人改了东西）；
    这份不会——它就是降级时唯一的兜底。
    """

    def __init__(self, redis):
        self.r = redis

    async def put(self, key: str, value: dict):
        await self.r.set(f"stale:{key}",
                         json.dumps({"v": value, "ts": time.time()}),
                         ex=7 * 86400)      # 留 7 天，够久了

    async def get(self, key: str, max_age: float = 3600) -> Stale | None:
        raw = await self.r.get(f"stale:{key}")
        if not raw:
            return None
        d = json.loads(raw)
        age = time.time() - d["ts"]
        if age > max_age:
            return None                 # 太旧了，宁可不给
        return Stale(value=d["v"], age=age, reason="upstream_unavailable")
```

**为什么这个设计比"返回 503"好**，用铺货场景举例：

- 客户要铺 500 个商品，要读每个商品的价格
- 上游挂了，但 3 分钟前的价格明明还有
- 返回 503 → 客户整个任务失败，重跑一遍
- 返回 `{stale: true, age: 180}` → 客户照常铺货，价格是 3 分钟前的，几乎没影响

而且 `stale: true` 和 `age` 这两个字段是**诚实的**。客户自己决定要不要接受——如果他做的是财务对账，他会拒绝；如果是铺货，他会接受。这个决定权应该在他手里，不该由你替他做。

> 一句话总结这个模块：**99.9% 的含义不是"永远返回正确数据"，是"永远返回可解释的结果"**。抛异常是"我不知道发生了什么"，返回 stale 数据是"我知道发生了什么，这是我能给的最好的"。

---

## 七、第 5 件事：主动探测 + 错误码归一

前四件事都在"出事之后"起作用。这一件是**在客户发现之前发现**。

### （1）探测：不要把客户的第一次请求当成探测

```python
# link/probe.py
import asyncio
import time

# 三平台 × 三类资源，每类一个固定样本
# 这三个样本必须提前准备好，且长期存在（找一个自己店铺的老商品）
PROBES = [
    ("taobao",  "item_get",  {"num_iid": "654320987654"}),
    ("taobao",  "trade_list", {"start": "2026-10-01 00:00:00"}),
    ("jd",      "item_get",  {"skuId": "100012043978"}),
    ("jd",      "order_list", {"startTime": "2026-10-01 00:00:00"}),
    ("alibaba", "item_get",  {"num_iid": "610947572360"}),
    ("alibaba", "order_list", {"startTime": "2026-10-01 00:00:00"}),
]


async def probe_once(gateways, metrics):
    for platform, api, params in PROBES:
        t0 = time.monotonic()
        ok, detail = True, "ok"
        try:
            await gateways[platform].call(
                api, params, budget_ms=8000, timeout=5.0,
            )
        except Exception as e:
            ok, detail = False, f"{type(e).__name__}: {e}"
        cost = (time.monotonic() - t0) * 1000

        metrics.observe(platform=platform, api=f"probe.{api}",
                        ok=ok, cost_ms=cost)

        if not ok:
            metrics.alert(f"[探测失败] {platform}/{api} | {detail} | {cost:.0f}ms")


async def probe_loop(gateways, metrics, interval: float = 30.0):
    while True:
        try:
            await probe_once(gateways, metrics)
        except Exception as e:                      # 探测器自己不能挂
            metrics.alert(f"[探测器异常] {e}")
        await asyncio.sleep(interval)
```

30 秒一次，三平台三类资源，一天 8640 次请求——很小，但它能提前发现三件**不主动查就永远发现不了**的事：

1. **授权过期**。token 到期的那一刻不会有人告诉你，客户第二天来投诉你才知道。
2. **平台改字段**。京东某次把 `wareQd` 改名了，字段还是返回、只是空了。如果你的解析不校验必填，数据就静默变空。探测里做一次字段非空校验就能抓到。
3. **限流额度变了**。平台调整配额从 100 QPS 降到 50 QPS，你不探测就只能等客户撞上去。

### （2）错误码归一：一行结构化日志

```python
# link/metrics.py
import json
import time


class Metrics:
    def __init__(self, sink):
        self.sink = sink

    def observe(self, *, platform: str, api: str, ok: bool,
                kind: str = "ok", cost_ms: float = 0.0,
                trace_id: str = "", retries: int = 0, stale: bool = False):
        # 一行 JSON，不要多行日志
        # 多行日志在并发下会串行错乱，grep 的时候想死
        self.sink.write(json.dumps({
            "ts": round(time.time(), 3),
            "trace": trace_id,
            "platform": platform,
            "api": api,
            "ok": ok,
            "kind": kind,          # ok / retry / slow / fatal / unknown
            "cost_ms": round(cost_ms, 1),
            "retries": retries,
            "stale": stale,
        }, ensure_ascii=False) + "\n")

    def alert(self, msg: str):
        self.sink.write(json.dumps(
            {"ts": round(time.time(), 3), "level": "alert", "msg": msg},
            ensure_ascii=False) + "\n")
```

`kind` 这个字段是全部监控的基石。有了它，你能直接算出：

- `fatal` 占比 → 文档和参数校验有没有问题
- `slow` 占比 → 配额够不够，是不是该提额了
- `retry` 占比 → **上游健康度**（这个指标比错误率更早预警）
- `stale` 占比 → 降级在多大范围内发生

> 一个经验值：**`retry` 占比超过 5% 就该警惕，超过 10% 说明上游已经不正常了**，即便此时业务成功率还是 100%。这个指标比可用性提前 10~30 分钟反应。

---

## 八、串起来：一个请求要过五道关

```python
# link/gateway.py
import asyncio
import time

from .budget import Budget
from .errors import Kind, classify
from .retry import backoff


class Gateway:
    """链路层门面。业务层只调这一个方法。"""

    def __init__(self, platform, client, bulkhead, breaker,
                 retry_budget, stale_cache, metrics):
        self.platform = platform
        self.client = client
        self.bulkhead = bulkhead
        self.breaker = breaker
        self.retry_budget = retry_budget
        self.stale = stale_cache
        self.metrics = metrics

    async def call(self, api: str, fn, *, budget_ms: int = 8000,
                   timeout: float = 6.0, cache_key: str | None = None,
                   idempotent: bool = True, trace_id: str = "", **kwargs):
        """
        fn: 适配器函数，签名 fn(timeout=..., **kwargs) -> dict
        """
        budget = Budget(budget_ms)
        t0 = time.monotonic()

        # 第 4 关：熔断（最先，最便宜）
        if not self.breaker.allow():
            self.metrics.observe(platform=self.platform, api=api, ok=False,
                                 kind="breaker_open", trace_id=trace_id)
            return await self._degrade(cache_key, "breaker_open")

        last_exc: Exception | None = None
        retries = 0

        for attempt in range(3):
            # 第 2 关：预算不够就别试了
            if not budget.enough(0.3):
                last_exc = last_exc or TimeoutError("budget exhausted")
                break

            # 第 3 关：全局重试预算
            if attempt > 0:
                if not self.retry_budget.allow():
                    break
                await asyncio.sleep(backoff(attempt - 1))
                retries += 1

            try:
                # 第 1 关：舱壁
                async with self.bulkhead:
                    result = await asyncio.wait_for(
                        fn(timeout=budget.slice(timeout), **kwargs),
                        timeout=budget.left(),
                    )
                self.breaker.record(True)
                self.metrics.observe(
                    platform=self.platform, api=api, ok=True,
                    cost_ms=(time.monotonic() - t0) * 1000,
                    trace_id=trace_id, retries=retries,
                )
                if cache_key:
                    await self.stale.put(cache_key, result)
                return result

            except Exception as e:
                kind = classify(self.platform, exc=e)
                self.breaker.record(False)
                last_exc = e

                self.metrics.observe(
                    platform=self.platform, api=api, ok=False,
                    kind=kind.value,
                    cost_ms=(time.monotonic() - t0) * 1000,
                    trace_id=trace_id, retries=retries,
                )

                # 参数错 / 签名错 / 没见过：重试是烧配额
                if kind in (Kind.FATAL, Kind.UNKNOWN):
                    break
                # 限流：多等一会儿，别急着再来一次
                if kind is Kind.SLOW:
                    await asyncio.sleep(1.0 + backoff(attempt))
                # 写接口：除非在适配器里带了幂等键，否则一次都不重试
                if not idempotent:
                    break

        # 第 5 关：降级
        stale = await self._degrade(cache_key, str(last_exc))
        if stale is not None:
            return stale
        raise last_exc

    async def _degrade(self, cache_key, reason):
        if not cache_key:
            return None
        s = await self.stale.get(cache_key)
        if s is None:
            return None
        self.metrics.observe(platform=self.platform, api="degrade",
                             ok=True, kind="stale", stale=True)
        return s.as_payload()
```

注意最后的 `_degrade`：**降级放在抛出异常之前**。这个顺序是有意的——只有"真的没招了"才降级，而不是一失败就降级（那样会把偶发抖动放大成大面积 stale 数据）。

---

## 九、五件事做完，数字变成了什么样

| 指标           | 做之前        | 做之后            | 主要来自哪件事     |
| ------------ | ---------- | -------------- | ----------- |
| 业务成功率        | 99.21%     | **99.94%**     | 五件叠加        |
| 单平台故障影响面     | 全站（三平台一起挂） | **该平台降级，其余正常** | ① 舱壁        |
| 故障时 P99 延迟   | 31s（重试到超时） | **1.2s**（快速失败） | ② 预算 + ④ 熔断 |
| 重试放大倍数       | 3.7×       | 1.12×          | ③ 重试预算      |
| 上游异常发现时间     | 客户投诉后（小时级） | **30 秒内**      | ⑤ 探测        |
| 深夜 P0 次数（半年） | 7          | 1              | ④ 熔断 + ⑤ 探测 |

最值得看的是第二行：**故障影响面从"全站"变成"一个平台"**。这件事比把 99.21 提到 99.94 更有价值——因为前者意味着"客户换个平台还能干活"，后者只是"数字好看一点"。

---

## 十、五个反直觉的结论

做完这一套，有几个发现跟直觉不太一样，可能比代码更有用：

**1. 加超时不等于更快，超时必须和重试联动。**  
单独设超时，只是把"永久卡住"变成"卡一会儿再卡一会儿"。真正起作用的是总预算——它管的不是单个请求，是整个业务动作的时间上限。

**2. 重试率超过 5%，重试就在让系统变慢。**  
因为重试是有成本的：占连接、占并发、占配额，还挤压正常请求。要监控**重试率**而不是重试次数——次数看着不多，比率可能已经失控了。

**3. 熔断打开容易，恢复难。**  
`half_open` 只放 1 个探测请求的话，那个请求只要恰好失败，就得再等一个完整冷却期。实测放 3 个、成功 2 个才恢复，恢复速度差好几倍。

**4. 连接池不是越大越好，超过限流额度就是纯浪费。**  
`max_connections` 给到 200，但平台限流 50 QPS，多出来的 150 个连接只是在网关排队，同时占着你的内存和文件句柄。连接数应该 ≈ `QPS × 平均耗时`，再留 20%。

**5. 最难的不是技术，是口径。**  
把"调用方参数错误"算进分母，可用性上限就是 94%，你把代码优化到极致也上不了 99.9%。先想清楚分母，再谈数字。

---

## 十一、最后

这五件事，本质上是同一件事的五个层面：**承认上游一定会出问题，然后把"出问题"的代价控制住**。

- 舱壁控制**范围**——只影响一个平台
- 预算控制**时间**——1 秒认输，不拖 30 秒
- 重试分级控制**放大**——不让一次失败变成十次请求
- 熔断降级控制**形态**——从"报错"变成"给可解释的旧数据"
- 探测归一控制**发现速度**——客户投诉之前就知道

单看每一件事都不复杂，加起来大概一千多行代码。但它们决定了你半夜能不能睡着。

我们把这套东西在三个平台上跑了两年，中间踩过的坑基本都在上面了（`keepalive_expiry` 那条尤其贵）。如果你正在自己接这三个平台，这套代码可以直接拿去改；如果你觉得这套活不值得自己养，那也正常——我们做的就是这件事。

代码里所有类都是独立可测的，`Breaker`、`RetryBudget`、`Budget` 这三个不依赖网络，建议先写单元测试再接入，别拿线上流量试。

---

*附：文中 `keepalive_expiry` 的推荐值取决于各平台网关的实际配置，25 秒是我们在淘宝/京东/1688 上的经验值，接入前建议先用 `curl -v` 观察 `Connection: close` 的时机再定。*