Python中如何实现异步的任务重试机制_利用tenacity库的异步重试装饰器-Python教程-PHP中文网

tenacity异步重试不能直接用@retry，因默认为同步装饰器，调用time.sleep()会阻塞事件循环；必须显式启用coro=True并确保目标函数为async def。

python中如何实现异步的任务重试机制_利用tenacity库的异步重试装饰器

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）：

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 对象，不是原始异常。

Python Hardener

通过完整的安全审查、代码整洁重构、健壮的错误处理、正确的日志记录以及完整的文档字符串覆盖，强化 Python 脚本。

下载

示例：记录每次重试前的异常类型和已尝试次数：

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