Python 重试库 Tenacity 核心用法与实战指南
在开发分布式系统、微服务或调用外部 API 时,网络抖动、服务超时或临时性故障是常见挑战。手动实现重试逻辑往往繁琐且难以维护。Tenacity 是一个强大且灵活的 Python 重试库,它提供了丰富的策略和钩子函数,帮助你优雅地处理不稳定操作。本文将深入讲解 Tenacity 的安装、配置、核心用法及高级特性,并提供完整的代码示例。
本文详细介绍 Python 重试库 Tenacity 的安装配置及核心功能。涵盖基础装饰器使用、停止策略、等待时间配置、异常类型过滤等常用选项。通过自定义重试条件、Jitter 随机延迟、基于返回值的重试逻辑等高级用法示例,展示如何构建健壮的重试机制。结合日志记录与最佳实践,帮助开发者有效处理网络波动与服务不可用问题,提升系统稳定性。

在开发分布式系统、微服务或调用外部 API 时,网络抖动、服务超时或临时性故障是常见挑战。手动实现重试逻辑往往繁琐且难以维护。Tenacity 是一个强大且灵活的 Python 重试库,它提供了丰富的策略和钩子函数,帮助你优雅地处理不稳定操作。本文将深入讲解 Tenacity 的安装、配置、核心用法及高级特性,并提供完整的代码示例。
使用 pip 安装 Tenacity:
pip install tenacity
Tenacity 的核心是通过 @retry 装饰器将重试逻辑应用到目标函数上。该装饰器允许你定义何时停止重试、重试之间的等待时间以及在何种条件下执行重试。
以下示例展示了如何配置最大重试次数:
from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3))
def do_something():
print("Doing something...")
raise Exception("Something went wrong!")
try:
do_something()
except Exception as e:
print(f"最终失败:{e}")
在此示例中,函数 do_something 每次调用都会抛出异常。Tenacity 会拦截异常并自动重试,直到达到 3 次尝试后停止,最后重新抛出原始异常。
Tenacity 的配置主要分为三个维度:停止策略(Stop)、等待策略(Wait)和重试条件(Retry)。
决定何时放弃重试。
stop_after_attempt(n): 尝试 n 次后停止。stop_after_delay(seconds): 总耗时超过指定秒数后停止。stop_any(*stoppers): 任意一个停止条件满足即停止。from tenacity import stop_after_attempt, stop_after_delay
@retry(stop=(stop_after_attempt(5) | stop_after_delay(10)))
def operation():
# 5 次尝试或 10 秒后停止
pass
决定两次重试之间的等待时间。
wait_fixed(n): 固定等待 n 秒。wait_exponential(multiplier=1, min=1, max=10): 指数退避,默认从 1 秒开始,每次翻倍,最大不超过 10 秒。这是生产环境最常用的策略。wait_random(min, max): 在最小和最大值之间随机等待,用于避免'惊群效应'。from tenacity import wait_exponential
@retry(wait=wait_exponential(multiplier=1, min=4, max=10))
def slow_operation():
pass
决定哪些情况需要重试。
retry_if_exception_type(exception): 仅当捕获到指定类型的异常时重试。retry_if_result(predicate): 根据函数的返回值决定是否重试。retry_unless_exception_type(exception): 除非捕获到指定异常才重试。from tenacity import retry_if_exception_type
@retry(retry=retry_if_exception_type(IOError))
def open_file(path):
# 仅 IOError 触发重试
pass
在重试过程中记录日志有助于排查问题。可以使用 before_sleep_log 或在 before_sleep 中自定义回调。
import logging
from tenacity import retry, wait_fixed, before_sleep_log
logger = logging.getLogger(__name__)
@retry(wait=wait_fixed(2), before_sleep=before_sleep_log(logger, logging.WARNING))
def risky_task():
raise ConnectionError("Network error")
有时函数没有抛出异常,但返回结果不符合预期(例如 API 返回了错误码)。此时可使用 retry_if_result。
from tenacity import retry, retry_if_result
def is_success(result):
return result.get('status') == 'OK'
@retry(retry=retry_if_result(lambda x: not is_success(x)), stop=stop_after_attempt(3))
def fetch_data():
return {'status': 'FAIL'}
在实际开发中,Tenacity 常用于封装 HTTP 客户端。
import requests
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10),
reraise=True
)
def get_url(url):
response = requests.get(url, timeout=5)
response.raise_for_status()
return response.json()
try:
data = get_url("https://api.example.com/data")
except Exception as e:
print(f"获取数据失败:{e}")
stop 策略,防止死循环消耗资源。wait_exponential,给服务端恢复时间。Tenacity 为 Python 应用提供了标准化的重试机制,极大地简化了容错代码的编写。通过灵活组合停止、等待和重试条件,开发者可以构建出适应不同网络环境和业务需求的稳健系统。掌握 Tenacity 的高级用法,能显著提升分布式系统的可用性。

微信公众号「极客日志」,在微信中扫描左侧二维码关注。展示文案:极客日志 zeeklog
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online
将 Markdown(GFM)转为 HTML 片段,浏览器内 marked 解析;与 HTML转Markdown 互为补充。 在线工具,Markdown转HTML在线工具,online
将 HTML 片段转为 GitHub Flavored Markdown,支持标题、列表、链接、代码块与表格等;浏览器内处理,可链接预填。 在线工具,HTML转Markdown在线工具,online
通过删除不必要的空白来缩小和压缩JSON。 在线工具,JSON 压缩在线工具,online