tenacity异步重试不能直接用@retry，因默认为同步装饰器，调用time.sleep()会阻塞事件循环；必须显式启用coro=True并确保目标函数为async def。
tenacity异步重试装饰器为什么不能直接用@retry？
因为默认的 @retry 是同步装饰器，它会在协程函数里阻塞事件循环——调用 time.sleep() 而非 asyncio.sleep()，导致整个异步流程卡死。你看到的现象通常是：程序“假死”、后续协程不执行、或抛出 RuntimeWarning: coroutine 'sleep' was never awaited。
必须显式启用异步支持：
- 使用
retrying.AsyncRetrying实例（底层方式） - 或更推荐：用
tenacity.retry配合wait=...和stop=...+coro=True参数（v8.0+） - 装饰的目标函数必须是
async def，否则会报TypeError: object ... can't be used in 'await' expression
如何正确配置一个带指数退避的异步重试？
关键不是堆参数，而是选对等待策略和停止条件。比如 HTTP 请求失败后，立刻重试没意义，但等 1s → 2s → 4s → 8s 更合理；同时得设上限，避免无限重试。
推荐写法（tenacity ≥ 8.2.0）：
立即学习“Python免费学习笔记（深入）”；
macOS UI 自动化技能包 —— 用 Python 控制你的 Mac
下载
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import aiohttp
import asyncio
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10),
retry=retry_if_exception_type((aiohttp.ClientError, asyncio.TimeoutError)),
coro=True
)
async def fetch_data(url):
async with aiohttp.ClientSession() as session:
async with session.get(url, timeout=5) as resp:
return await resp.text()
-
multiplier=1：首次等待 1s，第二次 2s，第三次 4s（指数增长） -
min=1, max=10：实际等待时间被截断在 1–10 秒之间，防止单次等待过长 -
coro=True是开关，缺了就退化成同步行为 -
retry_if_exception_type比retry_if_exception更轻量，避免每次异常都跑判断逻辑
自定义异步重试中的错误日志和监控怎么做？
tenacity 不自动打日志，但提供 before_sleep 钩子，这是插日志、埋点、告警的唯一可靠入口——注意它接收的是 retry_state 对象，不是原始异常。
示例：记录每次重试前的异常类型和已尝试次数：
import logging
logger = logging.getLogger(__name__)
def log_retry(retry_state):
exc = retry_state.outcome.exception()
logger.warning(
"Retrying %s, attempt %s/%s, due to %s",
retry_state.fn.__name__,
retry_state.attempt_number,
retry_state.retry_object.stop.max_attempt_number,
type(exc).__name__
)
@retry(
stop=stop_after_attempt(3),
wait=wait_fixed(2),
before_sleep=log_retry,
coro=True
)
async def unreliable_api_call():
...
- 不要在
before_sleep里 await 任何协程，它本身是同步函数 - 如果要发异步告警（如上报 Prometheus），得用线程池或后台任务，否则会阻塞重试流程
-
retry_state.outcome.exception()才是本次失败的真实异常；retry_state.outcome.result()在成功时才有值
遇到 RuntimeError: no running event loop 怎么办？
这通常发生在你把异步重试函数当成普通函数调用，比如在脚本顶层直接 fetch_data("https://...")，而没用 asyncio.run() 或 await。
- 错误写法：
result = fetch_data("https://example.com")→ 返回一个coroutine对象，没执行 - 正确写法：
result = await fetch_data("https://example.com")（在 async 函数内）或result = asyncio.run(fetch_data("https://example.com"))（顶层） - 若集成到 FastAPI/Starlette 中，直接
await即可；但别在__init__或类属性赋值里调用 - 测试时容易漏掉
asyncio.run()，建议用pytest-asyncio插件配合@pytest.mark.asyncio
异步重试真正的复杂点不在配置，而在于它把错误处理、等待调度、上下文生命周期全耦合在一起——少一个 coro=True，或错一次 await，表现都像“功能失效”，而不是明确报错。
