40.少有人知的 Python“重试机制”

Posted 2023-04-05

参考技术A 为了避免由于一些网络或其他不可控因素，而引起的功能性问题。比如在发送请求时，会因为网络不稳定，往往会有请求超时的问题。

这种情况下，我们通常会在代码中加入重试的代码。重试的代码本身不难实现，但如何写得优雅、易用，是我们要考虑的问题。

这里要给大家介绍的是一个第三方库 - Tenacity ，它实现了几乎我们可以使用到的所有重试场景，比如：

在使用它之前 ，先要安装它

无条件重试，重试之间无间隔

无条件重试，但是在重试之前要等待 2 秒

只重试7 次

重试 10 秒后不再重试

或者上面两个条件满足一个就结束重试

在出现特定错误/异常（比如请求超时）的情况下，再进行重试

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

如下示例，当 test_retry 函数返回值为 False 时，再进行重试

如果想对一个异常进行重试，但是最多重试3次。

下面这个代码是无效的，因为它会一直重试，重试三次的限制不会生效，因为它的条件是有顺序的，在前面的条件会先被走到，就永远走不到后面的条件。

如果你把 stop_after_attempt 写到前边，就没有问题了。

当出现异常后，tenacity 会进行重试，若重试后还是失败，默认情况下，往上抛出的异常会变成 RetryError，而不是最根本的原因。

因此可以加一个参数（ reraise=True ），使得当重试失败后，往外抛出的异常还是原来的那个。

当最后一次重试失败后，可以执行一个回调函数

输出如下

摘自黑魔法手册

Python之如何优雅的重试

在编码中，一些涉及网络连接的代码片段经常需要重试，本文讲解了如何一步一步实现一个优雅的retry装饰器以及tenacity库的使用。

---

原始版本v0.0

假如有一个函数形式如下，函数有一些建立网络连接的逻辑

def f():
    # do some connections
    return 0

为了避免偶尔的网络连接失败，需要加上重试机制，那么最简单的形式就是在对应的代码片段加一个循环，循环体里使用异常捕获，连接成功时退出循环，否则就重复执行相关逻辑，此时修改之后的函数f如下

def f():
    while 1:
        try:
            # do some connections
            break
        except ConnectionError:
            continue
    return 0

---

装饰器版本v1.0

可以使用装饰器对代码进行抽象。例如现在有两个函数f1和f2需要加上重试机制，写一个名为retry的装饰器函数，用其装饰f1和f2即可。这样做避免了对老代码的修改，同时也实现了代码复用。示例如下

def retry(f):
    def wrap(*args, **kwargs):
        while 1:
            try:
                return f(*args, **kwargs)
            except ConnectionError:
                continue
    return wrap

@retry()
def f1():
    # do some connections
    return 0

@retry()
def f2():
    # do some other connections
    return 0

---

带参数的装饰器版本v1.1

v1.0的版本retry装饰器还有一些问题，假如有的函数想重试3次，有的想重试5次，重试的间隔也根据不同函数不一样，v1.0是无法实现的。此时可以借助带参数的三层装饰器，例如以下代码实现的retry装饰器，可以传入times和interval两个参数来设定重试次数和重试间隔

def retry(times, interval):
    def decorator(f)
        def wrap(*args, **kwargs):
            while times:
                try:
                    return f(*args, **kwargs)
                except ConnectionError:
                    times -= 1
                    time.sleep(interval)
                    continue
        return wrap
    return decorator

# 重试3次每次间隔10秒
@retry(times=3, interval=10)
def f1():
    # do some connections
    return 0

# 重试5次每次间隔15秒
@retry(times=5, interval=15)
def f2():
    # do some other connections
    return 0

装饰器支持参数之后，可以根据需要定义更丰富的参数，比如通过参数来设定需要捕获哪些异常等。

---

tenacity版本

tenacity是一个第三方开源库，用于函数的重试，实际上它的功能与原理是上面自己写的代码类似的！只是它可定义的参数更丰富，如果不想重复造轮子，拿来直接用就可以。代码示例如下

from tenacity import retry, stop_after_attempt, wait_fixed

# 不带任何参数的重试
@retry
def f():
    # do some connections
    return 0

# 重试5次每次间隔15秒
@retry(stop=stop_after_attempt(5), wait=wait_fixed(15))
def f():
    # do some connections
    return 0