文章目录

  • 简介
  • 安装
  • Retrying
    • 初试
    • 最大重试次数
    • 最大重试时间
    • 重试间隔
    • 随机间隔
    • 指数级重试间隔
    • 自定义异常重试
    • 自定义返回重试
    • 参数
  • Tenacity
    • 初试
    • 最大重试次数
    • 最大重试时间
    • 组合停止条件
    • 重试间隔
    • 随机间隔
    • 指数级重试间隔
    • 重试间隔链
    • 自定义异常重试
    • 自定义返回重试
    • 显式重试
    • 重新捕获异常
    • 重试前记日志
    • 重试后记日志
    • 重试失败后记日志
    • 重试统计
    • 自定义回调函数
    • 上下文管理器
    • 支持异步和协程
    • 参数
  • 参考文献

简介

Retrying 是一个通用重试库，用于简化任何需要重试的任务，已不再维护，功能：

  • 通用装饰器
  • 指定停止条件，如重试次数
  • 指定等待条件，如重试间的指数回退休眠
  • 自定义异常重试
  • 自定义重试预期返回的结果

Tenacity 是上述库的分支，修复了一些BUG，增加了新功能：

  • 异步协程重试
  • 上下文管理器重试
  • 组合停止条件



推荐使用 Tenacity




安装

pip install retrying
pip install tenacity




Retrying

初试

重试无数次直至成功

import random
from retrying import retry

n = 0  # 记录重试次数


@retry
def do_something_unreliable():
    global n
    if random.randint(0, 10) > 1:
        n += 1
        raise Exception
    else:
        print(f'tried {n} times')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    # tried 1 times
    # tried 2 times
    # tried 4 times
    # tried 5 times
    # tried 13 times




最大重试次数

import random
from retrying import retry

n = 0  # 记录重试次数


@retry(stop_max_attempt_number=3)  # 最大重试次数
def do_something_unreliable():
    global n
    if random.randint(0, 10) > 1:
        n += 1
        raise Exception
    else:
        print(f'tried {n} times and success.')
        n = 0  # 重置重试次数


for i in range(5):
    try:
        do_something_unreliable()
    except Exception as e:
        print(f'tried {n} times but failed.')
        n = 0  # 重置重试次数
    # tried 3 times but failed.
    # tried 1 times and success.
    # tried 0 times and success.
    # tried 2 times and success.
    # tried 3 times but failed.




最大重试时间

单位为毫秒

import time
from retrying import retry

n = 0  # 记录重试次数


@retry(stop_max_delay=3000)  # 最大重试时间为3s
def do_something_unreliable():
    global n
    n += 1
    raise Exception


start = time.time()
try:
    do_something_unreliable()
except:
    pass
end = time.time()
print(f'tried {n} times, pass {end - start} seconds')
# tried 796060 times, pass 2.999951124191284 seconds




重试间隔

import time
import random
from retrying import retry

n = 0  # 记录重试次数
start = time.time()  # 开始时间


@retry(wait_fixed=500)  # 重试间隔0.5秒
def do_something_unreliable():
    global n, start
    if random.randint(0, 3) > 1:
        n += 1
        raise Exception
    else:
        end = time.time()
        print(f'tried {n} times, pass {end - start} seconds')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    start = time.time()  # 更新开始时间
    # tried 0 times, pass 0.0 seconds
    # tried 3 times, pass 1.538625955581665 seconds
    # tried 1 times, pass 0.5115864276885986 seconds
    # tried 1 times, pass 0.5024125576019287 seconds
    # tried 0 times, pass 0.0 seconds




随机间隔

import time
import random
from retrying import retry

n = 0  # 记录重试次数
start = time.time()  # 开始时间


@retry(wait_random_min=500, wait_random_max=1000)  # 重试间隔0.5-1秒
def do_something_unreliable():
    global n, start
    if random.randint(0, 3) > 1:
        n += 1
        raise Exception
    else:
        end = time.time()
        print(f'tried {n} times, pass {end - start} seconds')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    start = time.time()  # 更新开始时间
    # tried 1 times, pass 0.7865383625030518 seconds
    # tried 1 times, pass 0.5917379856109619 seconds
    # tried 6 times, pass 4.129276990890503 seconds
    # tried 0 times, pass 0.0 seconds
    # tried 3 times, pass 2.2903735637664795 seconds




指数级重试间隔

x为重试次数，如第一次重试间隔2s，第二次4s，第三次8s，第四次16s（不设封顶的话）

import time
import random
from retrying import retry

n = 0  # 记录重试次数
start = time.time()  # 开始时间


@retry(wait_exponential_multiplier=1000, wait_exponential_max=10000)  # 指数级重试间隔=2^x*1000ms，10s封顶
def do_something_unreliable():
    global n, start
    if random.randint(0, 3) > 1:
        n += 1
        raise Exception
    else:
        end = time.time()
        print(f'tried {n} times, pass {end - start} seconds')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    start = time.time()  # 更新开始时间
    # tried 0 times, pass 0.0 seconds
    # tried 1 times, pass 2.0003767013549805 seconds
    # tried 0 times, pass 0.0 seconds
    # tried 0 times, pass 0.0 seconds
    # tried 4 times, pass 24.02167558670044 seconds




自定义异常重试

只重试 IOError

import random
from retrying import retry

n = 0  # 记录重试次数


@retry(retry_on_exception=lambda x: isinstance(x, IOError))  # 自定义异常重试
def do_something_unreliable():
    global n
    if random.randint(0, 10) > 1:
        n += 1
        raise IOError
    else:
        print(f'tried {n} times')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    # tried 5 times
    # tried 2 times
    # tried 3 times
    # tried 6 times
    # tried 2 times

只重试 IOError，其余抛出

import random
from retrying import retry

n = 0  # 记录重试次数


@retry(retry_on_exception=lambda x: isinstance(x, IOError), wrap_exception=True)  # 自定义异常结果
def do_something_unreliable():
    global n
    if random.randint(0, 10) > 1:
        n += 1
        raise IOError
    else:
        print(f'tried {n} times')
        n = 0  # 重置重试次数
    if random.randint(0, 2) > 1:
        raise IndexError


for i in range(5):
    try:
        do_something_unreliable()
    except Exception as e:
        print(e)
    # tried 4 times
    # tried 5 times
    # tried 2 times
    # tried 6 times
    # tried 1 times
    # RetryError[Attempts: 2, Error:
    #   File "C:\Users\Administrator\Envs\test\lib\site-packages\retrying.py", line 200, in call
    #     attempt = Attempt(fn(*args, **kwargs), attempt_number, False)
    #   File "D:/mycode/xxx.py", line 18, in do_something_unreliable
    #     raise IndexError
    # ]




自定义返回重试

返回为 None 则重试

import random
from retrying import retry

n = 0  # 记录重试次数


@retry(retry_on_result=lambda x: x is None)  # 自定义异常重试
def do_something_unreliable():
    global n
    if random.randint(0, 10) > 1:
        n += 1
        return None
    else:
        return n


for i in range(5):
    print(f'tried {do_something_unreliable()} times')
    n = 0  # 重置重试次数
    # tried 10 times
    # tried 8 times
    # tried 0 times
    # tried 10 times
    # tried 3 times




参数

参数功能
无重试直至成功
stop_max_attempt_number最大重试次数
stop_max_delay最大重试时间（毫秒）
wait_fixed重试间隔（毫秒）
wait_random_min 和 wait_random_max随机重试间隔（毫秒）
wait_exponential_multiplier 和 wait_exponential_max指数级重试间隔（毫秒）
retry_on_exception自定义异常重试
wrap_exception是否抛出其余重试
retry_on_result自定义异常结果




Tenacity

初试

import random
from tenacity import retry

n = 0  # 记录重试次数


@retry
def do_something_unreliable():
    global n
    if random.randint(0, 10) > 1:
        n += 1
        raise Exception
    else:
        print(f'tried {n} times')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    # tried 1 times
    # tried 0 times
    # tried 4 times
    # tried 7 times
    # tried 3 times




最大重试次数

import random
from tenacity import retry, stop_after_attempt

n = 0  # 记录重试次数


@retry(stop=stop_after_attempt(3))  # 最大重试次数
def do_something_unreliable():
    global n
    if random.randint(0, 10) > 1:
        n += 1
        raise Exception
    else:
        print(f'tried {n} times and success.')
        n = 0  # 重置重试次数


for i in range(5):
    try:
        do_something_unreliable()
    except Exception as e:
        print(f'tried {n} times but failed.')
        n = 0  # 重置重试次数
    # tried 1 times and success.
    # tried 3 times but failed.
    # tried 3 times but failed.
    # tried 2 times and success.
    # tried 2 times and success.




最大重试时间

单位为秒

import time
from tenacity import retry, stop_after_delay

n = 0  # 记录重试次数


@retry(stop=stop_after_delay(3))  # 最大重试时间为3s
def do_something_unreliable():
    global n
    n += 1
    raise Exception


start = time.time()
try:
    do_something_unreliable()
except:
    pass
end = time.time()
print(f'tried {n} times, pass {end - start} seconds')
# tried 206456 times, pass 2.9916257858276367 seconds




组合停止条件

通过操作符 | 组合多个停止条件

import time
from tenacity import retry, stop_after_attempt, stop_after_delay

n = 0  # 记录重试次数


@retry(stop=stop_after_attempt(300000) | stop_after_delay(3))  # 最大重试30w次 或 最长重试3秒
def do_something_unreliable():
    global n
    n += 1
    raise Exception


start = time.time()
try:
    do_something_unreliable()
except:
    pass
end = time.time()
print(f'tried {n} times, pass {end - start} seconds')
# tried 206456 times, pass 2.9916257858276367 seconds




重试间隔

import time
import random
from tenacity import retry, wait_fixed

n = 0  # 记录重试次数
start = time.time()  # 开始时间


@retry(wait=wait_fixed(0.5))  # 重试间隔0.5秒
def do_something_unreliable():
    global n, start
    if random.randint(0, 3) > 1:
        n += 1
        raise Exception
    else:
        end = time.time()
        print(f'tried {n} times, pass {end - start} seconds')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    start = time.time()  # 更新开始时间
    # tried 1 times, pass 0.5027601718902588 seconds
    # tried 2 times, pass 1.0299296379089355 seconds
    # tried 0 times, pass 0.0 seconds
    # tried 3 times, pass 1.5316619873046875 seconds
    # tried 4 times, pass 2.0474536418914795 seconds




随机间隔

import time
import random
from tenacity import retry, wait_random

n = 0  # 记录重试次数
start = time.time()  # 开始时间


@retry(wait=wait_random(min=0.5, max=1))  # 重试间隔0.5-1秒
def do_something_unreliable():
    global n, start
    if random.randint(0, 3) > 1:
        n += 1
        raise Exception
    else:
        end = time.time()
        print(f'tried {n} times, pass {end - start} seconds')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    start = time.time()  # 更新开始时间
    # tried 3 times, pass 2.287365674972534 seconds
    # tried 0 times, pass 0.0 seconds
    # tried 2 times, pass 1.4969894886016846 seconds
    # tried 0 times, pass 0.0 seconds
    # tried 6 times, pass 4.51520299911499 seconds

同样可以组合（不过我觉得这样闲着蛋疼）

import time
import random
from tenacity import retry, wait_fixed, wait_random

n = 0  # 记录重试次数
start = time.time()  # 开始时间


@retry(wait=wait_fixed(0.5) + wait_random(min=0.5, max=1))  # 重试间隔1-1.5秒
def do_something_unreliable():
    global n, start
    if random.randint(0, 3) > 1:
        n += 1
        raise Exception
    else:
        end = time.time()
        print(f'tried {n} times, pass {end - start} seconds')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    start = time.time()  # 更新开始时间
    # tried 2 times, pass 2.9294729232788086 seconds
    # tried 0 times, pass 0.0 seconds
    # tried 0 times, pass 0.0 seconds
    # tried 3 times, pass 3.8608667850494385 seconds
    # tried 1 times, pass 1.4092319011688232 seconds




指数级重试间隔

x为重试次数，最小1s，最大10s

import time
import random
from tenacity import retry, wait_exponential

n = 0  # 记录重试次数
start = time.time()  # 开始时间


@retry(wait=wait_exponential(multiplier=1, min=1, max=10))  # 指数级重试间隔=2^x*1s，最小1s，最大10s
def do_something_unreliable():
    global n, start
    if random.randint(0, 3) > 1:
        n += 1
        raise Exception
    else:
        end = time.time()
        print(f'tried {n} times, pass {end - start} seconds')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    start = time.time()  # 更新开始时间
    # tried 0 times, pass 0.0 seconds
    # tried 6 times, pass 35.06491994857788 seconds
    # tried 2 times, pass 3.013124942779541 seconds
    # tried 1 times, pass 1.010573387145996 seconds
    # tried 0 times, pass 0.0 seconds




重试间隔链

import time
import random
from tenacity import retry, wait_fixed, wait_chain

n = 0  # 记录重试次数
start = time.time()  # 开始时间


@retry(wait=wait_chain(*[wait_fixed(0.5)] * 3 + [wait_fixed(1)] * 2 + [wait_fixed(2)]))  # 重试间隔链，先3个0.5秒，再2个1秒，之后都是2秒
def do_something_unreliable():
    global n
    if n == 10:
        return
    n += 1
    end = time.time()
    print(f'tried {n} times, pass {end - start} seconds')
    raise Exception


do_something_unreliable()
# tried 1 times, pass 0.0 seconds
# tried 2 times, pass 0.5056586265563965 seconds
# tried 3 times, pass 1.0193772315979004 seconds
# tried 4 times, pass 1.5333683490753174 seconds
# tried 5 times, pass 2.5386297702789307 seconds
# tried 6 times, pass 3.5489938259124756 seconds
# tried 7 times, pass 5.551833629608154 seconds
# tried 8 times, pass 7.559761047363281 seconds
# tried 9 times, pass 9.561469554901123 seconds
# tried 10 times, pass 11.570155143737793 seconds




自定义异常重试

只重试 IOError

import random
from tenacity import retry, retry_if_exception_type

n = 0  # 记录重试次数


@retry(retry=retry_if_exception_type(IOError))  # 自定义异常重试
def do_something_unreliable():
    global n
    if random.randint(0, 10) > 1:
        n += 1
        raise IOError
    else:
        print(f'tried {n} times')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    # tried 5 times
    # tried 2 times
    # tried 3 times
    # tried 6 times
    # tried 2 times




自定义返回重试

返回为 None 则重试

import random
from tenacity import retry, retry_if_result

n = 0  # 记录重试次数


@retry(retry=retry_if_result(lambda x: x is None))  # 自定义异常结果
def do_something_unreliable():
    global n
    if random.randint(0, 10) > 1:
        n += 1
        return None
    else:
        return n


for i in range(5):
    print(f'tried {do_something_unreliable()} times')
    n = 0  # 重置重试次数
    # tried 10 times
    # tried 8 times
    # tried 0 times
    # tried 10 times
    # tried 3 times




显式重试

from tenacity import retry, stop_after_delay, TryAgain


@retry(stop=stop_after_delay(3))
def do_something_unreliable(n):
    """显式重试"""
    n += 1
    if n == 10:
        raise TryAgain
    return n


n = 0
while n <= 15:
    try:
        n = do_something_unreliable(n)
    except:
        n += 1
    else:
        print(n, end=' ')
# 1 2 3 4 5 6 7 8 9 11 12 13 14 15 16




重新捕获异常

超过最大重试次数或时间后抛出异常

import time
from tenacity import retry, stop_after_attempt, stop_after_delay

n = 0
start = time.time()


@retry(reraise=True, stop=stop_after_attempt(3))
def do_something_unreliable():
    """超过最大重试次数后抛出异常"""
    global n
    n += 1
    raise IOError(f'tried {n} times but failed.')


@retry(reraise=True, stop=stop_after_delay(3))
def do_something_unreliable1():
    """超过最大重试时间后抛出异常"""
    end = time.time()
    raise IOError(f'tried {end - start:.4f} seconds but failed.')


try:
    do_something_unreliable()
except Exception as e:
    print(e)
try:
    do_something_unreliable1()
except Exception as e:
    print(e)
# tried 3 times but failed.
# tried 2.9864 seconds but failed.




重试前记日志

import sys
import logging
from tenacity import retry, stop_after_attempt, before_log

logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)

logger = logging.getLogger(__name__)


@retry(stop=stop_after_attempt(3), before=before_log(logger, logging.DEBUG))
def do_something_unreliable():
    raise IOError('Fail')


try:
    do_something_unreliable()
except Exception as e:
    print(e)
# RetryError[<Future at 0x1b0342d7a58 state=finished raised OSError>]
# DEBUG:__main__:Starting call to '__main__.do_something_unreliable', this is the 1st time calling it.
# DEBUG:__main__:Starting call to '__main__.do_something_unreliable', this is the 2nd time calling it.
# DEBUG:__main__:Starting call to '__main__.do_something_unreliable', this is the 3rd time calling it.




重试后记日志

import sys
import time
import logging
from tenacity import retry, stop_after_attempt, after_log

logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)

logger = logging.getLogger(__name__)


@retry(stop=stop_after_attempt(3), after=after_log(logger, logging.DEBUG))
def do_something_unreliable():
    time.sleep(0.5)
    raise IOError('Fail')


try:
    do_something_unreliable()
except Exception as e:
    print(e)
# DEBUG:__main__:Finished call to '__main__.do_something_unreliable' after 0.500(s), this was the 1st time calling it.
# DEBUG:__main__:Finished call to '__main__.do_something_unreliable' after 1.000(s), this was the 2nd time calling it.
# DEBUG:__main__:Finished call to '__main__.do_something_unreliable' after 1.500(s), this was the 3rd time calling it.
# RetryError[<Future at 0x22b45e07a58 state=finished raised OSError>]




重试失败后记日志

import sys
import time
import logging
from tenacity import retry, stop_after_attempt, before_sleep_log

logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)

logger = logging.getLogger(__name__)


@retry(stop=stop_after_attempt(3), before_sleep=before_sleep_log(logger, logging.DEBUG))
def do_something_unreliable():
    time.sleep(0.5)
    raise IOError('Fail')


try:
    do_something_unreliable()
except Exception as e:
    print(e)
# DEBUG:__main__:Retrying __main__.do_something_unreliable in 0.0 seconds as it raised OSError: Fail.
# DEBUG:__main__:Retrying __main__.do_something_unreliable in 0.0 seconds as it raised OSError: Fail.
# RetryError[<Future at 0x1c840b96ac8 state=finished raised OSError>]




重试统计

from tenacity import retry, stop_after_attempt


@retry(stop=stop_after_attempt(3))
def do_something_unreliable():
    raise IOError('Fail')


try:
    do_something_unreliable()
except:
    pass
print(do_something_unreliable.retry.statistics)
# {'start_time': 756545.281, 'attempt_number': 3, 'idle_for': 0, 'delay_since_first_attempt': 0.0}

do_something_unreliable.retry.statistics["attempt_number"] 可直接获取重试次数，不用手动计算




自定义回调函数

retry_state 是 RetryCallState 的实例

from tenacity import retry, stop_after_attempt, retry_if_result

n = 0


def return_value(retry_state):
    """自定义回调函数"""
    return retry_state.outcome.result()  # 返回函数产生的最后结果或异常


@retry(stop=stop_after_attempt(3),
       retry_error_callback=return_value,
       retry=retry_if_result(lambda x: isinstance(x, int)))
def do_something_unreliable():
    global n
    n += 1
    return n


print(do_something_unreliable())
# 3




上下文管理器

结合 for 循环和上下文管理器

from tenacity import Retrying, RetryError, stop_after_attempt

try:
    for attempt in Retrying(stop=stop_after_attempt(3)):
        with attempt:
            raise Exception('Fail')
except RetryError:
    pass

协程

from tenacity import AsyncRetrying, RetryError, stop_after_attempt


async def function():
    try:
        async for attempt in AsyncRetrying(stop=stop_after_attempt(3)):
            with attempt:
                raise Exception('Fail')
    except RetryError:
        pass




支持异步和协程

import trio
import asks
import tornado
from tenacity import retry


@retry
async def my_async_function(loop):
    await loop.getaddrinfo('8.8.8.8', 53)


@retry
@tornado.gen.coroutine
def my_async_function(http_client, url):
    yield http_client.fetch(url)


@retry(sleep=trio.sleep)
async def my_async_function(loop):
    await asks.get('https://example')




参数

参数功能取值
无重试直至成功
stop停止条件最大重试次数 stop_after_attempt
最大重试时间 stop_after_delay
wait等待条件重试间隔 wait_fixed
随机重试间隔wait_random
指数级重试间隔 wait_exponential
重试间隔链 wait_chain
retry重试条件自定义异常 retry_if_exception_type
自定义返回 retry_if_result
reraise是否抛出异常bool
before重试前动作重试前记日志 before_log
after重试后动作重试后记日志 after_log
before_sleep重试失败后动作before_sleep_log 重试失败后记日志
retry_error_callback回调函数




参考文献

  1. Retrying GitHub
  2. Tenacity GitHub
  3. Tenacity Documentation
  4. Tornado Documentation
  5. Trio Documentation
  6. asks Documentation
  7. 少有人知的 Python “重试机制”
  8. Python利器：retrying失败、异常重试模块库

更多推荐

Python重试库Retrying和Tenacity