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

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

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

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

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

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

一、先把 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，淘宝那侧开始超时，连接全部卡在淘宝上，连接池占满，京东的请求连连接都拿不到——京东明明好好的，也跟着挂。

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

# 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})")


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）超时按接口类型拆开，不是一刀切

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

# 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

用起来是这样：

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 件事：重试分级——别把所有错误都当"再试一次"

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

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

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

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


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


# ---- 平台业务码 → 归一 ----
_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% → 重试 → 请求量翻倍……重试本身成了压垮上游的那根稻草。这就是雪崩。

# 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 件事：熔断，加上"带标记的降级"

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

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

# 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()

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

# 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）探测：不要把客户的第一次请求当成探测

# 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）错误码归一：一行结构化日志

# 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 分钟反应。

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

# 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
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    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 更有价值——因为前者意味着"客户换个平台还能干活"，后者只是"数字好看一点"。

十、五个反直觉的结论

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

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 的时机再定。