为了避免由于一些网络或等其他不可控因素,而引起的功能性问题。比如在发送请求时,会因为网络不稳定,往往会有请求超时的问题。
这种情况下,我们通常会在代码中加入重试的代码。重试的代码本身不难实现,但如何写得优雅、易用,是我们要考虑的问题。
这里要给大家介绍的是一个第三方库 – Tenacity ,它实现了几乎我们可以使用到的所有重试场景,比如:

  1. 在什么情况下才进行重试?
  2. 重试几次呢?
  3. 重试多久后结束?
  4. 每次重试的间隔多长呢?
  5. 重试失败后的回调?

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

  1. $ pip install tenacity

anaconda安装:

  1. conda install -c conda-forge tenacity

简单例子

一个简单的重试功能,如果发生异常,则会一直重试,直到成功:
(1)无限重试

  1. @retry
  2. def never_give_up_never_surrender():
  3. print("Retry forever ignoring Exceptions, don't wait between retries") raise Exception

(2)重试指定的次数之后停止
如下重试7次后结束

  1. @retry(stop=stop_after_attempt(7)) def stop_after_7_attempts():
  2. print("Stopping after 7 attempts")
  3. raise Exception

(3)重试指定的一段时间
如下重试10秒,如果还失败就抛出异常结束

  1. @retry(stop=stop_after_delay(3)) def stop_after_10_s():
  2. print("Stopping after 10 seconds")
  3. raise Exception

(4)条件组合重试
如下任何一个条件满足,都会触发结束。

  1. @retry(stop=(stop_after_delay(10) | stop_after_attempt(5)))
  2. def stop_after_10_s_or_5_retries():
  3. print("Stopping after 10 seconds or 5 retries")
  4. raise Exception

(5)指定的时间间隔重试
如下每隔2秒重试一次

  1. @retry(wait=wait_fixed(2))
  2. def wait_2_s():
  3. print("Wait 2 second between retries")
  4. raise Exception

(6)随机的时间间隔重试
如下在1和2之间产生的随机数来重试。

  1. @retry(wait=wait_random(min=1, max=2))
  2. def wait_random_1_to_2_s():
  3. print("Randomly wait 1 to 2 seconds between retries")
  4. raise Exception

(7)指数级的间隔等待
通常用在分布式的服务里面

  1. @retry(wait=wait_exponential(multiplier=1, min=4, max=10))
  2. def wait_exponential_1():
  3. print("Wait 2^x * 1 second between each retry starting with 4 seconds, then up to 10 seconds, then 10 seconds afterwards")
  4. raise Exception

(8)固定的时间间隔加上随机的时间间隔

  1. @retry(wait=wait_fixed(3) + wait_random(0, 2))
  2. def wait_fixed_jitter():
  3. print("Wait at least 3 seconds, and add up to 2 seconds of random delay")
  4. raise Exception

(9)遇到指定的ERROR重试

  1. @retry(retry=retry_if_exception_type(IOError))
  2. def might_io_error():
  3. print("Retry forever with no wait if an IOError occurs, raise any other errors")
  4. raise Exception

(10)自定义校验返回值重试

  1. def is_none_p(value):
  2. return value is None
  3. @retry(retry=(retry_if_result(is_none_p) | retry_if_exception_type()))
  4. def might_return_none():
  5. print("Retry forever ignoring Exceptions with no wait if return value is None")