# 接口稳定率 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` 的时机再定。*
