Python retrying 库如何实现自动重试机制？

本文共计924个文字，预计阅读时间需要4分钟。

目录 + retrying 模块 + 一、简介 + 二、使用方法 + 1. 无参数 + 2. 有参数 + 2.1 stop_max_attempt_number + 2.2 stop_max_delay + 2.3 wait_fixed + 2.4 wait_random [min/max] + 2.5 wait_exponential [multiplier/max] + 2.6 其他可用参数 + retr

目录

• retrying 模块
  • 一、 简介
  • 二、 使用方法
    • 1、 无参数
    • 2、 有参数
      • 2.1 stop_max_attempt_number
      • 2.2 stop_max_delay
      • 2.3 wait_fixed
      • 2.4 wait_random_[min/max]
      • 2.5 wait_exponential_[multiplier/max]
      • 2.6 其他可用参数

retrying 模块 一、 简介

retrying是一个python的重试包，可以用来自动重试一些可能运行失败的程序段，retrying提供一个装饰器函数retry，被装饰的函数就会在运行失败的情况下重新执行，默认只要一直报错就会不断重试

安装：

pip install retrying 二、 使用方法 1、 无参数

使用 retry，默认的行为会一直重试， 没有时间等待，直到不再报错为止

from retrying import retry from random import randint @retry # 装饰器 def get_random(): int_r = randint(0, 100) if int_r > 0: print(f"该随机数等于{int_r}") raise IOError("该随机数大于0") else: return int_r print(f"该随机数等于{get_random()}") 2、 有参数 2.1 stop_max_attempt_number

stop_max_attempt_number： 用来设定最大的尝试次数，超过该次数就会停止

from retrying import retry from random import randint @retry(stop_max_attempt_number=1) def get_random(): int_r = randint(0, 100) if int_r > 0: print(f"该随机数等于{int_r}") raise IOError("该随机数大于0") else: return int_r print(f"该随机数等于{get_random()}") 2.2 stop_max_delay

stop_max_delay：从被装饰的函数开始执行的时间点开始到函数成功运行结束或失败报错中止的时间点，单位：毫秒

from retrying import retry from random import randint @retry(stop_max_delay=1) def get_random(): int_r = randint(0, 100) if int_r > 0: print(f"该随机数等于{int_r}") raise IOError("该随机数大于0") else: return int_r print(f"该随机数等于{get_random()}") 2.3 wait_fixed

wait_fixed：设置在两次retrying之间的停留时间

from retrying import retry from random import randint @retry(wait_fixed=1) def get_random(): int_r = randint(0, 100) if int_r > 0: print(f"该随机数等于{int_r}") raise IOError("该随机数大于0") else: return int_r print(f"该随机数等于{get_random()}") 2.4 wait_random_[min/max]

wait_random_[min/max]： 两次retrying之间的停留最短/最长时间

from retrying import retry from random import randint @retry(wait_random_min=1, wait_random_max=9) def get_random(): int_r = randint(0, 100) if int_r > 0: print(f"该随机数等于{int_r}") raise IOError("该随机数大于0") else: return int_r print(f"该随机数等于{get_random()}") 2.5 wait_exponential_[multiplier/max]

wait_exponential_[multiplier/max]：以指数的形式产生两次retrying之间的停留时间, 产生的值为2^previous_attempt_number * wait_exponential_multiplier, previous_attempt_number是前面已经retry的次数, 如果产生的这个值超过了wait_exponential_max的大小, 那么之后两个retrying之间的停留值都为wait_exponential_max

from retrying import retry from random import randint @retry(wait_exponential_multiplier=100, wait_exponential_max=10000) def get_random(): int_r = randint(0, 100) if int_r > 0: print(f"该随机数等于{int_r}") raise IOError("该随机数大于0") else: return int_r print(f"该随机数等于{get_random()}") 2.6 其他可用参数

def __init__(self, stop=None, wait=None, # 停止和等待时间 stop_max_attempt_number=None, # 最大尝试次数 stop_max_delay=None, # 最大运行时间 wait_fixed=None, # 延迟时间 wait_random_min=None, wait_random_max=None, # 随机等待时间 wait_incrementing_start=None, wait_incrementing_increment=None, # 运行下次，增加延时 wait_exponential_multiplier=None, wait_exponential_max=None, # 等待时间按指数形式增长 retry_on_exception=None, retry_on_result=None, wrap_exception=False, stop_func=None, # 每次完成后运行的新函数 wait_func=None, wait_jitter_max=None):

其他的我也没弄清楚，抱歉啊！！！