Python如何实现异常重试机制_使用tenacity装饰器配置指数退避策略-Python教程-PHP中文网

要触发指数退避，必须显式配置 wait=wait_exponential(multiplier=1, min=1, max=10)；仅用 @retry 默认是立即重试。

python如何实现异常重试机制_使用tenacity装饰器配置指数退避策略

tenacity 装饰器怎么写才能触发指数退避？

直接用 @retry 不会自动指数退避，必须显式配置等待策略。默认是立即重试（wait_none()），看起来像没重试一样。

关键是要传入 wait=wait_exponential()，它才是指数退避的核心。起始等待时间由 multiplier 控制（单位秒），最大上限用 max 限制，避免退避过长：

from tenacity import retry, stop_after_attempt, wait_exponential
<p>@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10)
)
def fetch_data():</p><h1>可能抛出 requests.exceptions.RequestException</h1><pre class="brush:php;toolbar:false;">return requests.get("https://www.php.cn/link/46b315dd44d174daf5617e22b3ac94ca").json()

multiplier=1：第一次重试前等 1 秒，第二次等 2 秒，第三次等 4 秒（2ⁿ⁻¹ × multiplier）
min=1 确保即使 multiplier 很小，也不低于 1 秒（防抖）
max=10 封顶，避免第 5 次重试等 16 秒 → 实际卡在 10 秒

哪些异常该重试？别让 ValueError 也重试

tenacity 默认重试所有异常，这很危险。网络超时可以重试，但 ValueError 或 TypeError 是代码逻辑问题，重试只会重复失败。

必须用 retry_if_exception_type 或 retry_if_exception 精确控制：

from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential
<p>@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10),
retry=retry_if_exception_type((
requests.exceptions.Timeout,
requests.exceptions.ConnectionError,
requests.exceptions.HTTPError
))
)
def fetch_data():
...
</p>

只对网络类异常重试，HTTPError 要看状态码——比如 5xx 可重试，4xx（如 404、400）通常不该重试
若需更细粒度判断，改用 retry_if_exception(lambda e: isinstance(e, HTTPError) and e.response.status_code >= 500)
漏配 retry=... 是常见疏忽，会导致非预期异常也被重试

重试时如何记录日志？别让失败悄无声息

tenacity 自带 before 和 after 钩子，但默认不输出任何信息。生产环境必须加日志，否则无法判断是否真在重试、重试了几次、最后一次错在哪。

macOS UI 自动化技能包 —— 用 Python 控制你的 Mac

让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限

下载

推荐用 before=before_log(logger, logging.DEBUG) + after=after_log(logger, logging.WARNING)：

import logging
from tenacity import before_log, after_log
<p>logger = logging.getLogger(<strong>name</strong>)</p><p>@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10),
retry=retry_if_exception_type(requests.exceptions.RequestException),
before=before_log(logger, logging.DEBUG),
after=after_log(logger, logging.WARNING)
)
def fetch_data():
...
</p>

before 在每次尝试前打 DEBUG 日志，含尝试序号
after 在每次尝试后打 WARNING（包括成功和失败），失败时会附带异常类型和消息
注意：如果最终仍失败，after 会记录最后一次失败；但异常本身不会被吞掉，仍向上抛出

异步函数也能用 tenacity 吗？得换 asyncio 版本

直接给 async def 函数加 @retry 会报错：RuntimeWarning: coroutine 'xxx' was never awaited。tenacity 默认不支持协程。

必须用 tenacity.AsyncRetrying 手动封装，或改用 @retry 的异步变体（v8.0+ 支持）：

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import asyncio
<p>@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10),
retry=retry_if_exception_type(asyncio.TimeoutError),
reraise=True
)
async def async_fetch():
async with aiohttp.ClientSession() as session:
async with session.get("<a href="https://www.php.cn/link/46b315dd44d174daf5617e22b3ac94ca">https://www.php.cn/link/46b315dd44d174daf5617e22b3ac94ca</a>") as resp:
return await resp.json()
</p>