Python重试模块retrying
本文内容纲要：
-Python重试模块retrying
Python重试模块retrying
工作中经常碰到的问题就是，某个方法出现了异常，重试几次。循环重复一个方法是很常见的。比如爬虫中的获取代理，对获取失败的情况进行重试。
刚开始搜的几个博客讲的有点问题，建议看官方文档，还有自己动手实验。
参考：
https://segmentfault.com/a/1190000004085023
https://pypi.org/project/retrying/
最初的版本
importrequests
classProxyUtil:
def__init__(self):
self._get_proxy_count=0
defget_proxies(self):
try:
r=requests.get('代理服务器地址')
#print('正在获取')
#raiseException("异常")
#print('获取到最新代理=%s'%r.text)
params=dict()
ifrandr.status_code==200:
proxy=str(r.content,encoding='utf-8')
params['http']='http://'+proxy
params['https']='https://'+proxy
else:
raiseException("获取代理失败,状态码%s"%(r.status_code))
returnparams
exceptException:
ifself._get_proxy_count<5:
print('第%d次获取代理失败，准备重试'%self._get_proxy_count)
self._get_proxy_count+=1
self.get_proxies()
else:
print('第%d次获取代理失败，退出'%self._get_proxy_count)
self._get_proxy_count=0
returndict()
if__name__=='__main__':
proxy=ProxyUtil()
proxy.get_proxies()
以上代码通过
try...except...捕获异常，并通过一个计数器判断获取代理的次数，获取失败递归调用自己，直到达到最大次数为止。
为了模拟失败，可以解开抛出异常的注释
下面来试试retrying模块
安装
pipinstallretrying
retrying提供一个装饰器函数retry，被装饰的函数会在运行失败的情况下重新执行，默认一直报错就一直重试。
importrequests
fromretryingimportretry
classProxyUtil:
def__init__(self):
self._get_proxy_count=0
@retry
defget_proxies(self):
r=requests.get('代理地址')
print('正在获取')
raiseException("异常")
print('获取到最新代理=%s'%r.text)
params=dict()
ifrandr.status_code==200:
proxy=str(r.content,encoding='utf-8')
params['http']='http://'+proxy
params['https']='https://'+proxy
if__name__=='__main__':
proxy=ProxyUtil()
proxy.get_proxies()
结果：
正在获取
正在获取
正在获取
...
正在获取(一直重复下去)
没有添加任何参数，默认情况下会一直重试，没有等待时间
#设置最大重试次数
@retry(stop_max_attempt_number=5)
defget_proxies(self):
r=requests.get('代理地址')
print('正在获取')
raiseException("异常")
print('获取到最新代理=%s'%r.text)
params=dict()
ifrandr.status_code==200:
proxy=str(r.content,encoding='utf-8')
params['http']='http://'+proxy
params['https']='https://'+proxy
#设置方法的最大延迟时间，默认为100毫秒(是执行这个方法重试的总时间)
@retry(stop_max_attempt_number=5,stop_max_delay=50)
#通过设置为50，我们会发现，任务并没有执行5次才结束！
#添加每次方法执行之间的等待时间
@retry(stop_max_attempt_number=5,wait_fixed=2000)
#随机的等待时间
@retry(stop_max_attempt_number=5,wait_random_min=100,wait_random_max=2000)
#每调用一次增加固定时长
@retry(stop_max_attempt_number=5,wait_incrementing_increment=1000)
#根据异常重试，先看个简单的例子
defretry_if_io_error(exception):
returnisinstance(exception,IOError)
@retry(retry_on_exception=retry_if_io_error)
defread_a_file():
withopen("file","r")asf:
returnf.read()
read_a_file函数如果抛出了异常，会去retry_on_exception指向的函数去判断返回的是True还是False，如果是True则运行指定的重试次数后，抛出异常，False的话直接抛出异常。
当时自己测试的时候网上一大堆抄来抄去的，意思是retry_on_exception指定一个函数，函数返回指定异常，会重试，不是异常会退出。真坑人啊！
来看看获取代理的应用(仅仅是为了测试retrying模块)
#定义一个函数用于判断返回的是否是IOError
defwraper(args):
returnisinstance(args,IOError)
classProxyUtil:
defget_proxies(self):
r=requests.get('http://47.98.163.40:17000/get?country=local')
print('正在获取')
raiseIOError
#raiseIndexError
print('获取到最新代理=%s'%r.text)
params=dict()
ifrandr.status_code==200:
proxy=str(r.content,encoding='utf-8')
params['http']='http://'+proxy
params['https']='https://'+proxy
#@retry_handler(retry_time=2,retry_interval=5,retry_on_exception=[IOError,IndexError])
@retry(stop_max_attempt_number=5,retry_on_exception=wraper)
defretry_test(self):
self.get_proxies()
print('io')
这种方法只能判断单一的异常，而且扩展性不够高
#通过返回值判断是否重试
defretry_if_result_none(result):
"""ReturnTrueifweshouldretry(inthiscasewhenresultisNone),Falseotherwise"""
#returnresultisNone
ifresult=="111":
returnTrue
@retry(stop_max_attempt_number=5,retry_on_result=retry_if_result_none)
defmight_return_none():
print("RetryforeverignoringExceptionswithnowaitifreturnvalueisNone")
return"111"
might_return_none()
might_return_none函数的返回值传递给retry_if_result_none的result，通过判断result,返回Treu或者None表示需要重试，重试结束后抛出RetryError，返回False表示不重试。
扩展默认的retry装饰器：
defretry_handler(retry_time:int,retry_interval:float,retry_on_exception:[BaseException],*args,**kwargs):
defis_exception(exception:[BaseException]):
forexpinretry_on_exception:
ifisinstance(exception,exp):
returnTrue
returnFalse
#returnisinstance(exception,retry_on_exception)
def_retry(*args,**kwargs):
returnRetrying(wait_fixed=retry_interval*1000).fixed_sleep(*args,**kwargs)
returnretry(
wait_func=_retry,
stop_max_attempt_number=retry_time,
retry_on_exception=is_exception
)
classProxyUtil:
defget_proxies(self):
r=requests.get('代理地址')
print('正在获取')
raiseIOError
#raiseIndexError
print('获取到最新代理=%s'%r.text)
params=dict()
ifrandr.status_code==200:
proxy=str(r.content,encoding='utf-8')
params['http']='http://'+proxy
params['https']='https://'+proxy
@retry_handler(retry_time=2,retry_interval=5,retry_on_exception=[IOError,IndexError])
#@retry(stop_max_attempt_number=5,retry_on_exception=wraper)
defretry_test(self):
self.get_proxies()
print('io')
if__name__=='__main__':
proxy=ProxyUtil()
proxy.retry_test()
本文内容总结：Python重试模块retrying，
原文链接：https://www.cnblogs.com/mangM/p/11207202.html