用 tenacity 替代手写重试逻辑,可避免重复造轮子、规避指数退避误配、统一控制重试边界,尤其适合中高并发或关键链路场景。

直接说结论:用 tenacity 替代手写 try/except + time.sleep 重试逻辑,能避免重复造轮子、规避指数退避误配、统一控制重试边界,尤其适合中高并发或关键链路的请求场景。
为什么 requests 自带重试不够用
requests 的 Retry 类(配合 HTTPAdapter)只覆盖基础 HTTP 状态码和连接类异常,对业务级失败(比如返回 500 但不是网络原因)、自定义错误判断、异步调用、或需要结合超时动态调整重试间隔的场景支持很弱。更常见的是:你写了重试,但没控制总耗时,结果三次重试叠加超时,反而比单次还慢;或者重试时没做 jitter,所有请求在第 2 秒同时涌向服务端,触发雪崩。
无序列表呈现典型短板:
-
Retry不支持按响应内容重试(例如只对{"code": "BUSY"}重试) - 无法与
timeout参数联动(比如“最多花 8 秒重试,超了就放弃”) - 不兼容
aiohttp或httpx异步客户端 - 重试间隔固定或仅靠
backoff_factor线性增长,缺乏 jitter 防抖能力
tenacity 的核心配置项怎么填
tenacity 把重试拆成「什么条件下重试」+「怎么重试」+「什么时候停」三部分,每个环节都可编程控制。最常用组合是:
立即学习“Python免费学习笔记(深入)”;
@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 次失败后等 1 秒,第 2 次失败后等 2 秒,第 3 次失败后等 4 秒……但被min/max截断 -
retry_if_exception_type(...):只对指定异常重试;若要加响应体判断,改用retry_if_result+ lambda - 必须显式传入
timeout到原始请求里,tenacity不接管网络层超时
如何把超时和重试真正串起来
很多人以为设了 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
<p>@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):</p><h1>每次尝试都用递减 timeout:第1次(3,10),第2次(2,6),第3次(1,3)</h1><pre class='brush:python;toolbar:false;'>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 装饰器——它会阻塞整个 event loop。必须用 tenacity.AsyncRetrying:
import asyncio from tenacity import AsyncRetrying, stop_after_attempt, wait_random_exponential import httpx <p>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()</p>
这里 wait_random_exponential 比纯指数多加了随机抖动(jitter),能有效打散重试时间点,避免请求洪峰。而 response.raise_for_status() 是关键:它把 HTTP 4xx/5xx 转成异常,才能被 tenacity 捕获并触发重试。
复杂点在于:异步重试函数不能直接当普通函数调用,必须用 await;且 AsyncRetrying 不支持装饰器语法,只能显式循环。


















