Python重试库Tenacity使用指南:配置、停止条件与回调详解

陆景行
陆景行
发布时间:2026-10-15 | 6人浏览过
原创

本文详细介绍Python第三方库Tenacity的高级重试功能。涵盖无条件重试、设置重试次数与时间限制、基于异常类型或结果的条件重试,以及重试失败后的错误重新抛出和回调函数配置。通过具体代码示例,帮助开发者优雅地处理网络请求超时等不可控因素。

介绍

为了应对网络波动等不可控因素导致的功能性故障,例如在发送HTTP请求时因网络不稳定引发的超时问题,我们需要在代码中引入重试逻辑。

虽然实现重试逻辑本身并不复杂,但如何编写出既优雅又易于使用的重试代码,是开发者需要重点考虑的问题。

本文推荐一个强大的第三方库——Tenacity(注:标题中的“重试机制”并非Python内置模块,而是指该库实现的重试功能)。它几乎覆盖了所有常见的重试场景,建议读者收藏、关注并点赞以获取更多技术分享。

例如:

在何种条件下触发重试?

最多允许重试多少次?

重试持续多久后终止?

每次重试之间的等待间隔是多少?

重试失败后如何执行回调函数?

在使用该库之前,首先需要安装它。

$ pip install tenacity

1. 最基本的重试方式

无条件重试,且重试之间没有等待间隔。

from tenacity import retry

@retry
def test_retry():
    print("等待重试,重试无间隔执行...")
    raise Exception

test_retry()

无条件重试,但在每次重试前等待2秒。

from tenacity import retry, wait_fixed

@retry(wait=wait_fixed(2))
def test_retry():
    print("等待重试...")
    raise Exception

test_retry()

2. 设置停止条件

限制最多重试7次。

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(7))
def test_retry():
    print("等待重试...")
    raise Exception

test_retry()

限制重试总时长为10秒,超过后不再重试。

from tenacity import retry, stop_after_delay

@retry(stop=stop_after_delay(10))
def test_retry():
    print("等待重试...")
    raise Exception

test_retry()

或者当上述两个条件中任意一个满足时,即终止重试。

from tenacity import retry, stop_after_delay, stop_after_attempt

@retry(stop=(stop_after_delay(10) | stop_after_attempt(7)))
def test_retry():
    print("等待重试...")
    raise Exception

test_retry()

3. 设置重试触发条件

仅在出现特定错误或异常(如请求超时)时才进行重试。

from requests import exceptions
from tenacity import retry, retry_if_exception_type

@retry(retry=retry_if_exception_type(exceptions.Timeout))
def test_retry():
    print("等待重试...")
    raise exceptions.Timeout

test_retry()

在满足自定义条件时才进行重试。

如下示例中,当test_retry函数返回值为False时,才触发重试。

from tenacity import retry, stop_after_attempt, retry_if_result

def is_false(value):
    return value is False

@retry(stop=stop_after_attempt(3),
       retry=retry_if_result(is_false))
def test_retry():
    return False

test_retry()

4. 重试失败后重新抛出原始异常

当出现异常时,Tenacity会执行重试。如果重试后依然失败,默认情况下,向上抛出的异常将变为RetryError,而非最根本的原因。

因此,可以添加参数reraise=True,使得当重试失败后,向外抛出的异常保持为原始异常。

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(7), reraise=True)
def test_retry():
    print("等待重试...")
    raise Exception

test_retry()

5. 设置回调函数

当最后一次重试失败后,可以执行一个指定的回调函数。

from tenacity import *

def return_last_value(retry_state):
    print("执行回调函数")
    return retry_state.outcome.result()  # 表示返回原函数的返回值

def is_false(value):
    return value is False

@retry(stop=stop_after_attempt(3),
       retry_error_callback=return_last_value,
       retry=retry_if_result(is_false))
def test_retry():
    print("等待重试中...")
    return False

print(test_retry())

输出结果如下

等待重试中... 等待重试中... 等待重试中... 执行回调函数 False

到此这篇关于Python重试库Tenacity使用指南的文章就介绍到这了,更多相关Python重试机制内容请搜索本站以前的文章或继续浏览下面的相关文章,希望大家以后多多支持本站!

本站声明

本文内容仅供参考,不代表本站观点。转载或引用请注明出处。如发现内容问题,请通过联系我们页面反馈。

Python 相关产品和应用

全部下载

专题列表