Python如何优雅地使用重试：tenacity

1 缘起

项目中使用了第三方服务，和上一篇文章一样：SpringBoot中如何优雅地使用重试https://blog.csdn.net/Xin_101/article/details/134617868
在调用第三方服务时，出现第三方服务连接不到的情况，为了保证服务的相对稳定，
当调用第三方服务出现异常时，进行重试，
本次的项目是Python语言开发的，因此，选择了灵活、功能强大的Tenacity框架。

2 Tenacity

https://tenacity.readthedocs.io/en/latest/

2.1 简介

Tenacity是Python的重试框架，用于简化重试逻辑的编写，
直接在Tenacity中提供的重试方法中配置相关参数即可实现重试功能，
参数化重试机制，如重试次数、时间间隔、重试的异常情况等，
提高开发效率。

2.2 retry方法

通过retry方法实现在指定异常情况下进行重试。
配置重试参数，在需要重试的方法上添加retry注解即可实现重试。
retry源码如下：
tenacity.retry

retry参数源码如下：

常用参数解析：

其中，

• max_retries是执行的总次数，重试次数即为执行的总次数-1，如max_retries=5，重试4次，总共执行5次。
• retry_exception用于配置重试的异常条件，即在何种异常出现时进行重试。
• reraise是重试后是否抛出异常的标识，配置为true时，当达到重试最大次数时，重新抛出异常，给下游捕获，当配置为false时，不抛出异常。
• min_seconds：最小时间间隔，当重试次数计算的时间间隔小于min_seconds时，使用min_seconds的值作为时间间隔，否则使用计算时间间隔。
• max_seconds：最大时间间隔，当重试次数计算的时间间隔大于max_seconds时，使用max_seconds作为时间间隔，避免延迟时间爆炸，限定重试间隔上限/上边界。
• multiplier用于配置重试时间间隔比率，时间间隔：multiplier * 2^(n-1),n为执行次数，这个可以参见源码，如下图所示，时间间隔为result=self.multiplier * exp，而exp=self.exp_base ** (retry_state.attempt_number - 1)，因此时间间隔为：multiplier * 2^(n-1)，最终的重试时间间隔是在计算结果result、最小时间间隔和最大时间间隔中取出的。
  位置：tenacity.wait.wait_exponential
  

2.3 配置retry参数

这里我们对retry方法进行一层包装，有些多余，
先这样写吧，如下：

2.4 指定方法上使用retry

完成retry方法参数配置后，在需要重试的方法中使用retry框架，
首先初始化retry的方法，
然后在执行的方法上添加注解。

执行结果如下图所示，
按照上面的retry配置，可知min_seconds=4s，max_seconds=10s，max_retries=5，multiplier=1
即重试4次，

• 时间间隔：第二次~第一次：time_interval=1*2^(2-1)=1s，time_interval<min_seconds，因此使用4s；
• 时间间隔：第三次~第二次：time_interval=1*2^(3-1)=4s，time_interval=min_seconds，因此使用4s；
• 时间间隔：第四次~第三次：time_interval=1*2^(4-1)=8s，min_seconds<time_interval<max_seconds，因此使用8s；

2.5 完整样例

"""
重试测试

@author xindaqi
@since 2023-12-02 16:34
"""
from typing import Callable, Any
from datetime import datetime
import logging


from tenacity import (
    before_sleep_log,
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential,
)

logger = logging.getLogger(__name__)


def retry_decorator(max_retries: int, retry_exception: "RetryBaseT" = retry_if_exception_type(), reraise=True, multiplier=1, min_seconds=4, max_seconds=10) -> Callable[[Any], Any]:
    """重试装饰器

    :param max_retries: 最大重试次数
    :param retry_exception: 触发重试的异常类型
    :param reraise: 重新抛出异常开关：true：重新抛出异常；false：不重新抛出异常
    :param multiplier: 重试时间倍率，时间间隔：multiplier * 2^(n-1),n为执行次数
    :param min_seconds: 最小重试时间间隔
    :param max_seconds: 最大重试时间间隔
    :return: 重试对象
    """
    return retry(
        reraise=reraise,
        stop=stop_after_attempt(max_retries),
        wait=wait_exponential(multiplier=multiplier, min=min_seconds, max=max_seconds),
        retry=retry_exception,
        before_sleep=before_sleep_log(logger, logging.WARNING),
    )


def function_with_retry(max_retries, retry_exception):
    retry_decorator_init = retry_decorator(max_retries, retry_exception)

    @retry_decorator_init
    def run():
        now = datetime.now()
        current_time = now.strftime("%Y-%m-%d %H:%M:%S")
        print(">>>>>>>>Current time:{}".format(current_time))
        a = 1/0
    return run()


if __name__ == "__main__":
    exception = (
        retry_if_exception_type(ZeroDivisionError))
    function_with_retry(5, exception)

3 小结

（1）Tenacity重试框架，提供重试参数配置，简单易用，提高开发效率；
（2）重试参数有max_retries（最大执行次数）、reraise（重新抛出异常标识，如果达到重试次数上限后，是否抛出原生异常给上游调用方），min_seconds最小的时间间隔，即重试的时间间隔最小只能为min_seconds，max_seconds是最大时间间隔，即最大的延迟时间为max_seconds，为重试时间间隔做了限制，保证间隔的合理；
（3）重试时间间隔计算公式为：time_interval=multiplier*2^(n-1)，n为重试次数，最终的延迟时间间隔从time_interval、min_seconds和max_seconds中取，计算公式：max(max(0, min_seconds), min(time_interval, max_seconds))；
（4）Tenacity可以配置多种异常类型，当出现这些异常时进行重试，细粒度控制重试。