工程优化- Tenacity库用于重试机制及使用示例

Tenacity 是一个用于 Python 的通用重试库，旨在简化在函数调用失败时自动重试的逻辑，提高程序的健壮性和容错能力。它特别适用于处理网络请求、数据库连接、外部 API 调用等可能因临时故障（如网络波动、服务暂时不可用）而失败的场景。

核心作用

1. 简化重试逻辑：通过装饰器或上下文管理器，可以轻松为任何函数添加可配置的重试行为，无需编写冗长的try-except循环。
2. 灵活的策略配置：
   • 停止条件：例如限制最大重试次数（stop_after_attempt）或总重试时间（stop_after_delay）。
   • 等待策略：支持固定间隔（wait_fixed）、指数退避（wait_exponential）、随机间隔等，以控制重试之间的延迟。
   • 重试条件：可以指定仅在特定异常类型（如IOError、ConnectionError）或自定义返回值条件（如返回False时）才触发重试。
3. 支持异步与回调：兼容异步代码（协程），并允许在重试前后执行自定义回调函数（如记录日志）。
4. 提升代码可读性：将重试逻辑与业务逻辑分离，使代码更清晰、易于维护。

安装

pipinstalltenacity

代码示例

以下是几个典型的使用场景示例。

示例 1：基本重试（网络请求）

在请求失败时最多重试 3 次，每次间隔 2 秒。

importrequestsfromtenacityimportretry,stop_after_attempt,wait_fixed@retry(stop=stop_after_attempt(3),wait=wait_fixed(2))deffetch_data(url):response=requests.get(url)response.raise_for_status()# 非 200 状态码会抛出异常returnresponse.json()# 使用try:data=fetch_data("https://api.example.com/data")print("成功获取数据:",data)exceptExceptionase:print(f"最终失败:{e}")

说明：如果请求抛出异常（如网络错误或 HTTP 错误），函数会自动重试，最多 3 次，每次等待 2 秒。

示例 2：指数退避等待

使用指数退避策略（首次等待 1 秒，后续按倍数增加，最多等待 10 秒），最多重试 5 次。

fromtenacityimportretry,stop_after_attempt,wait_exponential@retry(stop=stop_after_attempt(5),wait=wait_exponential(multiplier=1,min=1,max=10))defcall_external_api():# 模拟外部 API 调用response=requests.get("https://external.service/api")response.raise_for_status()returnresponse.text

说明：这种策略能避免对服务端造成瞬时压力，常用于分布式系统。

示例 3：仅针对特定异常重试

只在发生Timeout异常时重试，其他异常直接抛出。

fromrequestsimportexceptionsfromtenacityimportretry,retry_if_exception_type@retry(retry=retry_if_exception_type(exceptions.Timeout))defrequest_with_timeout():print("尝试请求...")raiseexceptions.Timeout# 模拟超时request_with_timeout()

说明：通过retry_if_exception_type可以精确控制重试的异常类型。

示例 4：根据返回值条件重试

当函数返回False时重试，最多 3 次。

fromtenacityimportretry,stop_after_attempt,retry_if_resultdefis_false(value):returnvalueisFalse@retry(stop=stop_after_attempt(3),retry=retry_if_result(is_false))defcheck_status():# 模拟检查状态，返回 False 表示未就绪returnFalsecheck_status()

说明：这种方式适用于需要根据结果（而非异常）决定是否重试的场景，例如等待某个条件达成。

示例 5：组合停止条件与重试后回调

设置“最多重试 5 次或总时间不超过 10 秒”的停止条件，并在重试失败后执行回调。

fromtenacityimportretry,stop_after_attempt,stop_after_delay,retry_if_exception_typedeflog_failure(retry_state):print(f"重试失败，最后一次异常:{retry_state.outcome.exception()}")@retry(stop=(stop_after_attempt(5)|stop_after_delay(10)),retry=retry_if_exception_type(IOError),retry_error_callback=log_failure)defread_file():withopen("temp.txt","r")asf:returnf.read()

说明：这里使用|组合多个停止条件，并在最终失败时通过回调记录日志。

注意事项

• 幂等性：确保重试的操作是幂等的（多次执行结果相同），避免因重试导致重复提交或数据不一致。
• 资源释放：在重试循环中注意资源（如连接、文件句柄）的及时释放。
• 超时设置：合理设置重试次数和等待时间，避免无限重试或过长延迟影响用户体验。

Tenacity 因其配置灵活、API 简洁，已成为 Python 生态中处理重试逻辑的首选库之一。通过上述示例，你可以快速将其集成到项目中，提升代码的可靠性。