Python如何优化网络请求超时_基于Tenacity的智能重试机制

先说结论：用 tenacity 替代手写 try/except + time.sleep 的重试逻辑，可以避免重复造轮子、规避指数退避的误配问题，还能统一控制重试边界。这套方案尤其适合中高并发场景，或者那些对请求成功率有严格要求的核心链路。

你可能会问，难道 requests 自带的 Retry 不够用吗？其实，Retry 类（配合 HTTPAdapter）只能覆盖基础的 HTTP 状态码和连接类异常。一旦遇到业务层面的失败——比如服务端返回了 500，但问题并非出在网络层面——或者需要自定义重试判断逻辑、适配异步调用，甚至是根据超时动态调整重试间隔，它就显得力不从心了。

更常见的痛点在于：你明明写了重试逻辑，却没控制总耗时。结果三次重试叠加超时，反而比单次请求更慢。要么就是重试时没有加入随机抖动（jitter），所有请求在某一秒同时涌向服务端，直接触发雪崩效应。

总结一下，requests.Retry 的主要短板包括：

• 不支持按响应内容进行重试，比如只对返回 {"code": "BUSY"} 的情况重试
• 无法与 timeout 参数联动，比如“最多花 8 秒来重试，超时就直接放弃”
• 不兼容 aiohttp 或 httpx 这类异步客户端
• 重试间隔要么固定，要么仅靠 backoff_factor 线性增长，缺少 jitter 防抖能力

tenacity 的核心配置项怎么填

tenacity 把重试逻辑拆解为三个独立环节：什么条件下重试、怎么重试、什么时候停止。每个环节都可以通过编程灵活控制。最常见的一个组合配置如下：

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type(requests.exceptions.Timeout)
)

这里几个关键参数值得留意：

• stop_after_attempt(3) 表示最多尝试 3 次（包含首次请求），注意不是“再试 3 次”
• wait_exponential(...) 实现指数退避：第一次失败后等待 1 秒，第二次失败后等待 2 秒，第三次失败后等待 4 秒……以此类推，但会被 min 和 max 参数截断
• retry_if_exception_type(...) 表示只对指定的异常类型进行重试。如果还需要检查响应体内容，就必须改用 retry_if_result 配合 lambda 函数来判断返回值
• 需要特别注意的是，tenacity 并不接管网络层的超时，必须显式地在原始请求中传入 timeout 参数

如何把超时和重试真正串起来

很多人以为设置了 timeout=(3, 10) 再套上 tenacity 就万事大吉，其实这里漏掉了一个关键约束：重试的总耗时很可能远超预期。举个例子，连接超时 3 秒 × 3 次 = 9 秒，加上读取超时 10 秒 × 3 次 = 30 秒，实际最长可能卡住 39 秒。

正确的做法是使用 stop_after_delay(8) 来强制设定总时间上限，同时让每次请求的 timeout 动态缩短：

from tenacity import retry, stop_after_attempt, stop_after_delay, wait_fixed, retry_if_exception_type
import requests

@retry(
    stop=(stop_after_attempt(3) | stop_after_delay(8)),  # 任一条件满足即停止重试
    wait=wait_fixed(1),
    retry=retry_if_exception_type((requests.exceptions.Timeout, requests.exceptions.ConnectionError))
)
def fetch_with_limited_total_time(url):
    # 每次尝试都使用递减的 timeout: 第1次(3,10)，第2次(2,6)，第3次(1,3)
    attempt = fetch_with_limited_total_time.retry.statistics.get("attempt_number", 1)
    timeouts = {1: (3, 10), 2: (2, 6), 3: (1, 3)}
    timeout = timeouts.get(attempt, (1, 3))
    return requests.get(url, timeout=timeout)

这里要特别提醒一点：retry_if_exception_type 只捕获异常，无法处理 response.status_code == 503 这类“请求成功但业务失败”的情况。如果需要处理这类场景，就得改用 retry_if_result 加上自定义的判断函数。

容易被忽略的异步场景适配

如果你用的是 aiohttp 或 httpx.AsyncClient，千万别直接把同步版的 tenacity 装饰器套上去——那样会阻塞整个事件循环。正确的做法是使用 tenacity.AsyncRetrying：

import asyncio
from tenacity import AsyncRetrying, stop_after_attempt, wait_random_exponential
import httpx

async def fetch_async(url):
    async for attempt in AsyncRetrying(
        stop=stop_after_attempt(3),
        wait=wait_random_exponential(multiplier=1, min=1, max=5)
    ):
        with attempt:
            async with httpx.AsyncClient() as client:
                response = await client.get(url, timeout=5)
                response.raise_for_status()
                return response.json()

在这个例子中，wait_random_exponential 比纯指数退避多了一个随机抖动（jitter）机制，能有效打散重试的时间点，避免形成请求洪峰。而 response.raise_for_status() 是关键的转折点：它会把 HTTP 4xx/5xx 状态码转化为异常，这样 tenacity 才能捕获到并触发重试。

需要留意的是：异步重试函数不能像普通函数那样直接调用，必须用 await。同时 AsyncRetrying 不支持装饰器语法，只能通过显式的 async for 循环来实现。