Tenacity异步支持好吗

wen python案例 2026-07-05 305

本文目录导读：

Tenacity异步支持好吗

核心支持：async 修饰器
关键特性：异步等待不阻塞
与其他异步组件的兼容性
高级功能：异步的停止/等待/重试条件
一个常见的误区：第三方库需要是异步的
总结与建议

Tenacity 对异步（async/await）的支持非常好，可以说是目前 Python 中用于重试逻辑的库中，对异步支持最完善、最成熟的一个。

如果你在写 asyncio 代码（比如使用 aiohttp, httpx, asyncpg 等库），Tenacity 可以让你用几乎和同步代码完全一样的方式来编写重试逻辑。

下面从几个关键方面详细说明：

核心支持：`async` 修饰器

Tenacity 的核心装饰器 @tenacity.retry 原生支持异步函数，你只需要把装饰器加在 async def 函数上，它就能正确 await 你的函数并处理其异常。

示例：

import asyncio
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
async def fetch_data(url: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        response.raise_for_status()
        return response.json()
# 使用
async def main():
    try:
        data = await fetch_data("https://api.example.com/data")
        print(data)
    except Exception:
        print("最终失败")
asyncio.run(main())

这段代码会：

尝试执行 fetch_data。
如果抛出异常（如 httpx.HTTPStatusError），它会根据 wait_exponential 策略异步等待（不会阻塞事件循环）。
最多重试 3 次，重试间隔呈指数增长（2s, 4s, 8s）。
3 次都失败，最终异常会向上抛出。

关键特性：异步等待不阻塞

这是最核心的优势,当 Tenacity 需要等待下次重试时，它使用的是 asyncio.sleep() 而不是 time.sleep()，这意味着：

事件循环不会被阻塞：在等待重试的这段时间里，事件循环可以去处理其他协程（处理其他请求）。
高效：不会因为等待而浪费 CPU 或 I/O 资源。

与其他异步组件的兼容性

Tenacity 可以很好地与各种异步框架和库结合使用：

Web 框架：在 FastAPI, Sanic, aiohttp 的路由处理函数中使用。
数据库：对 asyncpg, aiomysql, motor (MongoDB) 等异步数据库客户端进行重试。
消息队列：如 aio-pika (RabbitMQ)。
AI 客户端：如 openai 库的异步客户端。

高级功能：异步的停止/等待/重试条件

Tenacity 不仅支持异步函数，其回调函数也可以定义为异步的，这意味着你可以编写：

异步的 before_sleep：在重试前异步记录日志或发送通知。
异步的 retry_state_callback：根据异步结果（比如检查状态码、调用另一个 API）来决定是否重试。
异步的 retry_error_callback：最终失败时，进行异步的清理或降级操作。

示例：异步的日志回调

import asyncio
import logging
from tenacity import retry, stop_after_attempt, before_sleep
logger = logging.getLogger(__name__)
async def async_before_sleep(retry_state):
    # 异步记录日志，例如写入数据库或发送请求
    await asyncio.sleep(0)  # 模拟异步操作
    logger.warning(f"重试 #{retry_state.attempt_number}, 等待 {retry_state.next_action.sleep} 秒, 异常: {retry_state.outcome.exception()}")
@retry(stop=stop_after_attempt(3), before_sleep=async_before_sleep)
async def my_async_func():
    # ... 可能失败的异步操作
    raise ValueError("Something went wrong")

一个常见的误区：第三方库需要是异步的

非常重要的一点：Tenacity 解决的是 “如何重试” 的问题，它不能把 同步函数变成异步函数。

如果你重试的是一个同步、阻塞的函数（requests.get），那么整个重试过程（包括内部等待）仍然是同步且阻塞事件循环的。
正确的用法是：你有一个原生异步的函数（如 httpx.AsyncClient.get, aiohttp.ClientSession.get），然后用 Tenacity 装饰它。

总结与建议

特性	评价
核心理念	原生支持 `async/await`，API 与同步版本一致。
等待机制	使用 `asyncio.sleep()`，不阻塞事件循环。
兼容性	与 `aiohttp`, `httpx`, `FastAPI`, `asyncpg` 等主流异步库无缝集成。
高级功能	支持异步的回调函数（`before_sleep`, `retry_state_callback` 等）。
学习成本	低，如果你用过同步的 Tenacity，几乎可以直接迁移。
性能	在异步场景下性能很好，没有不必要的同步阻塞。

如果你的项目基于 asyncio，Tenacity 是进行重试操作的最佳选择，它设计优雅，功能强大，且没有异步编程中常见的“阻塞事件循环”的陷阱。
如果你还在用 requests 等同步库，建议先升级到其异步版本（如 httpx），然后再结合 Tenacity 使用，这样才能发挥其异步支持的全部优势。