在编写Python代码进行自动化测试、网络爬虫或者其他与网络相关的动作的时候,由于网络影响会容易失败,而这种失败并不是我们需要去处理的。那么这种时候最好的办法就是失败后重试几次,以避免网络的间断性影响。
如果我们正常编写代码的话,可能需要 try...except ,但是这种写法很麻烦,能实现的效果也很单一。这里介绍一个 Python 库retrying
,专门用来对抛出异常的函数或者方法进行重试。
通过 retrying
你能干什么:
- 出错后重新运行函数,直到正常运行为止;
- 出错后,暂停一会再运行函数,因为网络可能一时半会不会好;
- 出错后,如果重试时间过长,会造成代码效率过低,你可以设置一个最大的重试时间;
- 出错时,你可能想去执行另一个函数以排除可能的错误。
那接下来我们看看用法,首先你需要安装这个库:
pip install retrying
简单用法
简单的示例,我们写一个函数,其中有一个变量 a 从 1 到 2 中随机取一个, 当 a 为 1 的时候就抛出异常,我们尝试在抛出异常时会不会重试:
import random
def demo():
a = random.randint(1,2)
print(a) # => 先打印,好看效果
if a == 1:
raise # 为1时抛出异常
这个函数有 50% 的概率会抛出错误,你可以先试试。不过如果你运气好的话。。。
好,接下来,我们加上 retrying :
from retrying import retry # => 引入retry装饰器
import random
@retry() # => 装饰你想重试的函数,注意这里有括号
def demo():
a = random.randint(1,2)
print(a) # => 先打印,好看效果
if a == 1:
raise # 为1时抛出异常
demo() # => 运行这个可能报错的函数
运行几次看看,是不是当打印 a 为 1 的时候会直到 a 为 2 为止?也就意味着@rery()
处理了异常并重试。
更多参数
也许你不满足于让它默认重试,那么我们来看看更多的函数。
重试次数
stop_max_attempt_number
最大重试次数,默认为5次
设定一个最大重试次数,超过这个次数会停止重试,并把异常抛出来。按照你的需要设置吧,一般不要太多就行。或者你也可以不设,默认为 5 次,不过前提时你修复了下面的bug。
import random
from retrying import retry
@retry(stop_max_attempt_number=3)
def demo():
a = random.randint(1,2)
print(a, end=" ") # 为了排版,打印横排
if a != 3: # => 这里我们改为会始终抛异常的情况,不可能取到3
raise
demo()
## 1 1 1
## RuntimeError: No active exception to reraise
我们可以看到,运行 3 次都报错后,第 3 次运行的异常就抛出了。
这里源代码有个 bug,如果不设这个参数会无限重试(如果你的异常不会中止就是死循环)。我们来修复这个 bug 😋:
打开 Python安装目录\Lib\site-packages\retrying.py
,找到第87
行
if stop_max_attempt_number is not None:
# 改为
if self._stop_max_attempt_number is not None:
时间相关
wait_fixed
重试的间隔时间
当函数抛出异常后,下一次重试会间隔wait_fixed
设置的时间。默认是 1000 毫秒(1秒),你可以通过这个参数修改这个默认值。
wait_random_min
和 wait_random_max
用来设置随机的间隔时间
wait_random_min 和 wait_random_max 搭配起来设置默认的随机等待时间,默认是 0 ~ 1000 毫秒之间随机等待。
wait_incrementing_increment
每重试一次,持续增加等待时间
默认是 100 毫秒 ,每重试一次,等待时常就会增加 100 毫秒。
stop_max_delay
最长重试延迟时间,单位毫秒
这个不是重试间隔时间,这是函数运行 + 重试结束的整体时间。比如stop_max_delay=2000
也就是说该函数从运行到重试结束为止的时间为 2000 毫秒,超过就结束重试并抛出异常。如果你的函数运行时间已经超过 stop_max_delay 时间,就并不会重试。
import time
import random
from retrying import retry
@retry(stop_max_delay=2000) # => 设置了最大的重试时间
def demo():
a = random.randint(1, 2)
print(a, end=" ")
time.sleep(1) # => 这里等待了1秒
if a != 3:
raise
demo()
## 1 2
## RuntimeError: No active exception to reraise
关联函数重试
retry_on_result
:指定一个函数,如果指定的函数返回True,则重试,否则抛出异常退出。
import random
from retrying import retry
def fun():
# 处理代码
return True # => 返回True
@retry(retry_on_result=fun) # 指定函数,如果fun返回True则重试,否则不重试
def demo():
a = random.randint(1, 2)
print(a, end=" ")
if a != 3:
raise
demo()
shop_func
:指定被装饰函数出错后,会执行的函数,执行该函数后在来重试被装饰的函数。注意,该函数必须要有两个参数(attempts, delay)。用来当抛出异常后,需要做一些处理的时候。
a = random.randint(1, 2)
def fun(attempts, delay):
global a
a = 3
print('待测函数')
@retry(stop_func=fun)
def demo():
print(a, end=" ")
time.sleep(1)
if a != 3:
raise
demo()