python中可以发生异常自动重试库retrying

发布时间：2024-11-23 15:01

1、前言

在我们日常的开发工作中，为了代码的健壮性，我们通常会对方法中的业务代码进行try-except的包装，以防止在发生异常的时候进程被中断。如果发生异常，我们该怎么办呢？有同学可能用自己的方式（循环）去做了重试，那么有没有一种通用的可靠的重试方式呢？答案是有的，它就是retrying库，今天我们就一起来看看。

2、快速开始

retrying是一个极简的使用Python编写的，拥有1.8k⭐️的可以实现方法异常重试的库，我们在实际使用中直接为我们的方法增加一个@retry的装饰器即可。

2.1、安装retrying

pip install retrying

2.2、模拟无限重试

假设我们现在有一个方法demo1，这个方法中调用了另一个接口e1，但是被调用的这个e1接口不稳定，当e1接口返回异常的时候，我们需要去重试。

from retrying import retry
import time

# 一个会返回异常的接口e1
def e1():
    time.sleep(1)
    print(\'err\')
    # 我们抛出异常模拟接口异常
    raise TypeError

# 全局计数器
a=1
@retry
def demo1(n):
    # 方法中调用全局变量，需要global
    global a
    # 进行try-except
    try:
        print(f\'开始尝试！{a}\')
        a+=1
        e1()
    except Exception as e:
        print(e)
        # 当重试完成后还未成功，则返回超时
        raise TimeoutError
if __name__==\'__main__\':
    demo1(0)

代码执行结果：

$\"python中可以发生异常自动重试库retrying_第1张图片\"$

你会发现，我们代码会一值被重试，这是为什么呢？

默认的retry装饰器就是无限重试的，直到重试成功为止。因为我们的接口e1是永远返回异常的，所以这个重试将会永远持续下去。

我们改造e1方法如下，即可在重试第10次成功：

# 一个会返回异常的接口e1
def e1():
    time.sleep(1)
    print(\'err\')
    # 当重试次数达到10时，返回True
    if a>=10:
        print(\'ok!\')
        return True
    # 我们抛出异常模拟接口异常
    raise TypeError

代码执行结果：

开始尝试！1
err
开始尝试！2
err
开始尝试！3
err
开始尝试！4
err
开始尝试！5
err
开始尝试！6
err
开始尝试！7
err
开始尝试！8
err
开始尝试！9
err
ok!

2.3、模拟最大重试次数

上面的无限重试明显在实际业务场景中适用面不是非常广，当我们需要使用重试次数来限制时，我们就可以使用retry的装饰器参数stop_max_attempt_number来实现。

# 全局计数器
a=1
@retry(stop_max_attempt_number=3)
def demo1(n):
    # 方法中调用全局变量，需要global
    global a
    # 进行try-except
    try:
        print(f\'开始尝试！{a}\')
        a+=1
        e1()
    except Exception as e:
        print(e)
        # 当重试完成后还未成功，则返回超时
        raise TimeoutError

以上代码表示，我们只重试3次，如果还未成功，则返回超时异常。

代码执行结果：

$\"python中可以发生异常自动重试库retrying_第2张图片\"$

2.4、模拟最大重试时间

除了可以使用重试次数限制外，当我们对时间要求比较明确的时候，我们就可以使用stop_max_delay参数来指定最大重试时间。这个参数的单位是ms。

# 全局计数器
a=1
@retry(stop_max_delay=2000)
def demo1(n):
    # 方法中调用全局变量，需要global
    global a
    # 进行try-except
    try:
        print(f\'开始尝试！{a}\')
        a+=1
        e1()
    except Exception as e:
        print(e)
        # 当重试完成后还未成功，则返回超时
        raise TimeoutError

如上代码表示，当重试时间超过2s就会抛出超时异常。因为在e1方法中，每次调用都会休眠1s，所以对于限制两秒，我们只能在2s内重试2次。

代码执行结果：

$\"python中可以发生异常自动重试库retrying_第3张图片\"$

2.5、模拟重试间隔时间

当我们不希望重试间隔时间太长的时候，我们可以使用参数wait_fixed指定重试的间隔时间。

# 全局计数器
a=1
@retry(stop_max_delay=6000,wait_fixed=1000)
def demo1(n):
    # 方法中调用全局变量，需要global
    global a
    # 进行try-except
    try:
        print(f\'开始尝试！{a}\')
        a+=1
        e1()
    except Exception as e:
        print(e)
        # 当重试完成后还未成功，则返回超时
        raise TimeoutError

如上代码表示最大重试时间6秒，重试间隔1秒，但是e1方法中还休眠了1秒，所以在6秒内，我们理论上只能重试4次（不是3次）。

代码运行结果：

$\"python中可以发生异常自动重试库retrying_第4张图片\"$

在此基础上，retrying还为我们提供了随机的重试间隔时间参数。

wait_random_min，重试间隔最小时间
wait_random_max，重试间隔最大时间

一般，这两个参数都是成对出现，用于限制重试间隔时间的范围。

2.6、指定重试调用的方法

retrying允许我们在重试的同时去调用一个方法。使用参数stop_func来指定这个方法名。

# attempts, delay这两个参数是必填的
def stop_f(attempts, delay):
    print(\'发生异常了，正在进行重试！\')
# 全局计数器
a=1
@retry(stop_func=stop_f,stop_max_delay=3)
def demo1(n):
    # 方法中调用全局变量，需要global
    global a
    # 进行try-except
    try:
        print(f\'开始尝试！{a}\')
        a+=1
        e1()
    except Exception as e:
        print(e)
        # 当重试完成后还未成功，则返回超时
        raise TimeoutError

代码运行结果：

$\"python中可以发生异常自动重试库retrying_第5张图片\"$

但是stop_func存在一个问题就是和stop_max_delay、stop_max_attempt_number结合使用的时候，后面两者会失效。

2.7、指定重试的异常类型

当我们需要针对特定的异常才进行重试时，就需要用到这个参数：retry_on_exception。

2.8、指定重试的特定条件

当达到某个条件才会进行重试，需要使用参数：retry_on_result。

到此这篇关于python中可以发生异常自动重试库retrying的文章就介绍到这了,更多相关python retrying内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

python中可以发生异常自动重试库retrying

目录

1、前言

2、快速开始

2.1、安装retrying

2.2、模拟无限重试

2.3、模拟最大重试次数

2.4、模拟最大重试时间

2.5、模拟重试间隔时间

2.6、指定重试调用的方法

2.7、指定重试的异常类型

2.8、指定重试的特定条件

相关推荐