tenacity是python生产环境最稳妥的重试方案,因其支持业务逻辑错误判断、自定义条件、日志可观测及与异步/熔断等机制协同,而requests.retry仅限网络层容错。

tenacity 是目前 Python 生产环境中最稳妥的重试方案,它比手写 while 循环或 requests 的 retry adapter 更可控、更可观测,也更容易与异步、超时、熔断等机制配合。
为什么不用 requests.adapters.Retry?
requests 内置的 Retry 对象只作用于 HTTP 状态码和连接异常,对业务逻辑错误(如返回 {"code": 401, "msg": "token expired"})、解析失败、数据缺失等完全无感。它也不支持自定义重试条件、无法嵌入日志上下文、难以调试重试路径。
常见误用现象:max_retries=3 设了但实际没重试——因为响应是 200,只是内容为空;或者重试了但每次都在用过期 token,越重试越错。
实操建议:
- 仅在纯网络层容错(DNS 失败、ConnectionError、5xx)时,才用
requests.adapters.Retry配合HTTPAdapter - 业务级重试必须上
tenacity,把“是否重试”的判断权交还给你自己 - 不要把
tenacity和Retry混着用——容易触发双重重试,导致请求量翻倍
@retry 装饰器怎么写才不踩坑?
最简可用写法是 @retry(stop=stop_after_attempt(3)),但线上出问题时你会发现:重试了 3 次,日志里只有一条成功记录,失败过程全黑盒。
实操建议:
- 必加
before=before_log(logger, logging.DEBUG)和after=after_log(logger, logging.DEBUG),否则根本不知道哪次失败、为什么失败 - 避免直接装饰整个爬虫函数;应拆到最小原子操作,比如单独装饰
fetch_page()和parse_json(),便于定位是网络问题还是解析问题 - 慎用
wait_exponential():默认从 1 秒开始翻倍,3 次重试最长等 4 秒,对高并发采集可能拖慢整体吞吐;建议显式设multiplier=0.5或改用wait_fixed(1)
示例(关键部分):
from tenacity import retry, stop_after_attempt, wait_fixed, before_log, after_log
import logging
logger = logging.getLogger(__name__)
@retry(
stop=stop_after_attempt(3),
wait=wait_fixed(1),
before=before_log(logger, logging.DEBUG),
after=after_log(logger, logging.DEBUG)
)
def fetch_page(url: str) -> dict:
resp = requests.get(url, timeout=10)
resp.raise_for_status()
return resp.json()
如何让重试感知 token 过期并自动刷新?
这是业务重试最典型的耦合点:重试不是简单地“再发一次”,而是“换凭证再发”。tenacity 本身不处理状态,但提供了 retry_error_callback 和 retry 的上下文钩子。
实操建议:
- 在
retry装饰器外维护一个可变的auth_token变量,或封装成类实例属性 - 用
retry_error_callback捕获最后一次失败的RetryError,检查其.last_attempt.result()或.last_attempt.exception()是否含 token 相关提示 - 更推荐方式:在被装饰函数内部主动 raise 自定义异常(如
TokenExpiredError),然后用retry_if_exception_type(TokenExpiredError)触发重试,并在重试前调用refresh_token()
注意:refresh_token() 本身也得加 @retry,且重试策略要更激进(比如最多 2 次,固定等待 0.3 秒),避免卡死在认证环节。
异步爬虫(aiohttp)能用 tenacity 吗?
能,但必须用 tenacity.AsyncRetrying,且装饰器不能直接套 async def 函数——得手动 await。
常见错误现象:写了 @retry 但协程没 await,结果返回 coroutine object,后续 await 时报 RuntimeWarning: coroutine 'xxx' was never awaited。
实操建议:
- 别用装饰器,改用上下文管理器模式:
from tenacity import AsyncRetrying, stop_after_attempt, wait_fixed
async for attempt in AsyncRetrying(
stop=stop_after_attempt(3),
wait=wait_fixed(1)
):
with attempt:
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
return await resp.json()
这种写法明确控制执行流,也方便在每次尝试前后插日志或更新 header。
tenacity 的真正难点不在语法,而在于重试边界的划分——哪些该重试、哪些该报错、哪些该跳过。一个没想清楚的 retry_if_result(lambda x: not x) 可能让空列表永远重试下去。先理清你的失败分类,再配策略,比急着加装饰器重要得多。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!











