HTTPX 异步客户端重试传输层

示例说明

把重试做进传输层的好处是对上层完全透明：业务代码照常发请求，传输层负责瞬态故障的自愈。判定逻辑双管齐下——方法幂等性与响应状态码共同决定是否重试；限流响应优先读取服务端建议的等待时长，没有才用本地指数退避，并对上限封顶防止退避爆炸。

技术要点

• 传输层重试对业务代码透明
• Retry-After 头优先于本地退避算法
• 随机抖动防止客户端同步重试

代码解读

RetryingTransport 继承 AsyncHTTPTransport 并覆写请求处理：循环里先算退避再发请求，可重试状态码且方法幂等时 continue 进入下一轮，传输错误被捕获继续循环；耗尽次数后抛出带最后一次错误详情的异常。main 演示把自定义传输注入客户端。

使用建议

写操作重试必须配幂等键。重试次数与超时预算要统筹，防止总耗时超过上游容忍。连续重试失败应熔断而非无限循环。

代码示例

import asyncio
import random

import httpx


class RetryingTransport(httpx.AsyncHTTPTransport):
    def __init__(self, retries: int = 3, **kwargs):
        super().__init__(**kwargs)
        self._retries = retries

    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
        retryable = request.method in {'GET', 'HEAD', 'OPTIONS'}
        last_error: Exception | None = None

        for attempt in range(self._retries + 1):
            if attempt > 0:
                backoff = min(2 ** attempt * 0.5, 8.0)
                await asyncio.sleep(backoff + random.uniform(0, 0.3))
            try:
                response = await super().handle_async_request(request)
                if response.status_code in {429, 502, 503, 504} and retryable:
                    retry_after = response.headers.get('Retry-After')
                    if retry_after and retry_after.isdigit():
                        await asyncio.sleep(int(retry_after))
                    continue
                return response
            except (httpx.TransportError,) as exc:
                last_error = exc
        raise httpx.TransportError(f'exhausted retries: {last_error}')


async def main() -> None:
    transport = RetryingTransport(retries=3)
    async with httpx.AsyncClient(transport=transport, timeout=10.0) as client:
        resp = await client.get('https://api.example.com/status')
        print('final status:', resp.status_code)


asyncio.run(main())