python重试组件tenacity介绍

我是攻城师

发布于 2019-03-06 15:12:49

1.9K0

发布于 2019-03-06 15:12:49

文章被收录于专栏：我是攻城师

前言

在开发python项目时，不可避免的会用到一些重试功能，比如数据库和网络重连，或者其他的一些异常方法重试等等，有些组件可能自带了重试功能，但有些组件可能没有带就需要我们自己开发了，不过这种组件一般都有开源成熟的方案，所以我们就没必要重新造轮子了，而tenacity就是python里面一款功能强大的重试组件，活跃程度较高，支持python2和python3。

github地址：

https://github.com/jd/tenacity

安装方式

pip安装：

pip install tenacity

anaconda安装：

conda install -c conda-forge tenacity

简单例子

一个简单的重试功能，如果发生异常，则会一直重试，直到成功：

（1）无限重试

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

（2）重试指定的次数之后停止

如下重试7次后结束

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

（3）重试指定的一段时间

如下重试10秒，如果还失败就抛出异常结束

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

（4）条件组合重试

如下任何一个条件满足，都会触发结束。

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

（5）指定的时间间隔重试

如下每隔2秒重试一次

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

（6）随机的时间间隔重试

如下在1和2之间产生的随机数来重试。

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

（7）指数级的间隔等待

通常用在分布式的服务里面

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

（8）固定的时间间隔加上随机的时间间隔

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

（9）遇到指定的ERROR重试

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

（10）自定义校验返回值重试

def is_none_p(value):
    """Return True if value is None"""
    return value is None

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

正常结束的程序返回值是None，如果是则重试。

（11）多种组合重试

def is_ok(value):
    return value is False  # When value is False, it will retry

@retry(
retry=retry_if_result(is_ok) ,
wait=wait_fixed(2),
stop=stop_after_attempt(4)
)
def put(msg):
    try:
        send(msg)
        return True
    except:
        logger.exception("send error!")
        return False

如果结果是False就执行重试，重试的间隔是2秒，重试的次数是4

更多例子可参考：

https://tenacity.readthedocs.io/en/latest/

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2019-01-21，如有侵权请联系 cloudcommunity@tencent.com 删除

python