Python重试装饰器，手把手教你实现

一、前语

在写事务代码时分，有许多场景需求重试某块事务逻辑，例如网络请求、购物下单等，希望产生反常的时分多重试几回。本文分享如何运用Python 的装修器来进行面向切面（AOP）的方法实现自动重试器。

二、简略剖析

一个重试装修器，最重要的便是产生意外反常处理失利自动重试，有如下几点需求留意

• 失利不能一直重试，由于或许会出现死循环浪费资源，因而需求有 最大重试次数 或者 最大超时时刻

• 不能重试太频频，由于太频频简略导致重试次数很快用完，却没有成功呼应，需求有 重试时刻距离 来限制，有时能够加大成功概率，例如网络请求时有一段时刻是堵塞的，或者对方服务负载太高导致一段时刻无法呼应等。

简略剖析完，咱们的重试装修器，就要支撑可装备最大重试次数、最大超时时刻、重试距离，所以装修器就要设计成带参数装修器。

三、代码模仿实现

重试装修器-初版

剖析完毕后，看看第一版的装修器

import time
from functools import wraps
def task_retry(max_retry_count: int = 5, time_interval: int = 2):
    """
    使命重试装修器
    Args:
        max_retry_count: 最大重试次数 默许5次
        time_interval: 每次重试距离 默许2s
    """
    def _task_retry(task_func):
        @wraps(task_func)
        def wrapper(*args, **kwargs):
            # 函数循环重试
            for retry_count in range(max_retry_count):
                print(f"execute count {retry_count + 1}")
                try:
                    task_result = task_func(*args, **kwargs)
                    return task_result
                except Exception as e:
                    print(f"fail {str(e)}")
                    time.sleep(time_interval)
        return wrapper
    return _task_retry

装修器内部闭包，就简略经过 for 循环 来履行指定重试次数，成功获取结果就直接 return 回来，产生反常则睡眠装备重试距离时刻后继续循环

写个例子来模仿测验下看看作用

@task_retry()
def user_place_order():
    print("user place order success")
    return {"code": 0, "msg": "ok"}
ret = user_place_order()
print(ret)
>>>out
execute count 1
user place order success
{'code': 0, 'msg': 'ok'}

没有反常正常履行，在函数中模仿一个反常来进行重试看看

@task_retry(max_retry_count=3, time_interval=1)
def user_place_order():
    a = 1 / 0
    print("user place order success")
    return {"code": 0, "msg": "ok"}
ret = user_place_order()
print("user place order ret", ret)
>>>out
fail division by zero
execute count 2
fail division by zero
execute count 3
fail division by zero
user place order ret None

能够看到 user_place_order 函数履行了三遍，都产生了除零反常，最后超越最大履行次数，回来了 None 值，咱们能够在主逻辑中来判断回来值是否为 None 来进行超越最大重试次数失利的事务逻辑处理

ret = user_place_order()
print("user place order ret", ret)
if not ret:
    print("user place order failed")
    ...

重试装修器-改进版

现在只能装备 最大重试次数 没有最大超时时刻，有时分咱们想不光有重试，还得在规定时刻内完结，不想浪费太多试错时刻。所以添加一个 最大超时时刻装备选项默许为None，有值时超越最大超时时刻退出重试。

def task_retry(max_retry_count: int = 5, time_interval: int = 2, max_timeout: int = None):
    """
    使命重试装修器
    Args:
        max_retry_count: 最大重试次数 默许 5 次
        time_interval: 每次重试距离 默许 2s
        max_timeout: 最大超时时刻，单位s 默许为 None,
    """
    def _task_retry(task_func):
        @wraps(task_func)
        def wrapper(*args, **kwargs):
            # 函数循环重试
            start_time = time.time()
            for retry_count in range(max_retry_count):
                print(f"execute count {retry_count + 1}")
                use_time = time.time() - start_time
                if max_timeout and use_time > max_timeout:
                    # 超出最大超时时刻
                    print(f"execute timeout, use time {use_time}s, max timeout {max_timeout}")
                    return
                try:
                    return task_func(*args, **kwargs)
                except Exception as e:
                    print(f"fail {str(e)}")
                    time.sleep(time_interval)
        return wrapper
    return _task_retry

看看作用

# 超时
@task_retry(max_retry_count=3, time_interval=1, max_timeout=2)
def user_place_order():
    a = 1 / 0
    print("user place order success")
    return {"code": 0, "msg": "ok"}
>>>out
execute count 1
fail division by zero
execute count 2
fail division by zero
execute count 3
execute timeout, use time 2.010528802871704s, max timeout 2
user place order ret None
# 超越最大重试次数
@task_retry(max_retry_count=3, time_interval=1)
def user_place_order():
    a = 1 / 0
    print("user place order success")
    return {"code": 0, "msg": "ok"}
>>>out
execute count 1
fail division by zero
execute count 2
fail division by zero
execute count 3
fail division by zero
user place order ret None
# 正常
@task_retry(max_retry_count=3, time_interval=1, max_timeout=2)
def user_place_order():
    # a = 1 / 0
    print("user place order success")
    return {"code": 0, "msg": "ok"}
>>>out
execute count 1
user place order success
user place order ret {'code': 0, 'msg': 'ok'}

重试装修器-加强版

到这重试装修器基本功能就实现了，但还能够加强，Python现在支撑 async 异步方法写法，因而要是能够兼容异步写法那就更好了。先看看装修异步函数会是什么样的作用

import time
import asyncio
import functools
def task_retry(max_retry_count: int = 5, time_interval: int = 2, max_timeout: int = None):
    """
    使命重试装修器
    Args:
        max_retry_count: 最大重试次数 默许 5 次
        time_interval: 每次重试距离 默许 2s
        max_timeout: 最大超时时刻，单位s 默许为 None,
    """
    def _task_retry(task_func):
        @wraps(task_func)
        def wrapper(*args, **kwargs):
            # 函数循环重试
            start_time = time.time()
            for retry_count in range(max_retry_count):
                print(f"execute count {retry_count + 1}")
                use_time = time.time() - start_time
                if max_timeout and use_time > max_timeout:
                    # 超出最大超时时刻
                    print(f"execute timeout, use time {use_time}s, max timeout {max_timeout}")
                    return
                try:
                    return task_func(*args, **kwargs)
                except Exception as e:
                    print(f"fail {str(e)}")
                    time.sleep(time_interval)
        return wrapper
    return _task_retry
@task_retry(max_retry_count=3, time_interval=1, max_timeout=2)
def user_place_order():
    # a = 1 / 0
    print("user place order success")
    return {"code": 0, "msg": "ok"}
@task_retry(max_retry_count=3, time_interval=2, max_timeout=5)
async def user_place_order_async():
    """异步函数重试事例"""
    a = 1 / 0
    print("user place order success")
    return {"code": 0, "msg": "ok"}
async def main():
    # 同步事例
    # ret = user_place_order()
    # print(f"user place order ret {ret}")
    # 异步事例
    ret = await user_place_order_async()
    print(f"user place order ret {ret}")
if __name__ == '__main__':
    asyncio.run(main())
# 正常时分
execute count 1
user place order success
user place order ret {'code': 0, 'msg': 'ok'}
# 反常时分
>>>out
execute count 1
Traceback (most recent call last):
  File "G:/code/python/py-tools/decorator/base.py", line 138, in <module>
    asyncio.run(main())
  File "G:\softs\DevEnv\python-3.7.9\lib\asyncio\runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "G:\softs\DevEnv\python-3.7.9\lib\asyncio\base_events.py", line 587, in run_until_complete
    return future.result()
  File "G:/code/python/py-tools/decorator/base.py", line 133, in main
    ret = await user_place_order_async()
  File "G:/code/python/py-tools/decorator/base.py", line 121, in user_place_order_async
    a = 1 / 0
ZeroDivisionError: division by zero
Process finished with exit code 1

发现产生反常的时分并没有重试，为什么呢？其实在履行 task_func() 它并没有真实的履行内部逻辑，而是回来一个 coroutine 协程目标，并不会报反常，所以再装修器中履行一遍就成功就出来了，外面 ret = await user_place_order_async()， 后才真实的等待履行，然后履行函数内的逻辑再报反常就没有捕获到。咱们能够打断点验证下

这样装修器就不支撑异步函数的重试，需求加强它，能够运用 asyncio.iscoroutinefunction() 来进行异步函数的判断， 然后再加一个异步函数的闭包就能够实现异步、同步函数都兼容的重试装修器。

def task_retry(max_retry_count: int = 5, time_interval: int = 2, max_timeout: int = None):
    """
    使命重试装修器
    Args:
        max_retry_count: 最大重试次数 默许 5 次
        time_interval: 每次重试距离 默许 2s
        max_timeout: 最大超时时刻，单位s 默许为 None,
    """
    def _task_retry(task_func):
        @functools.wraps(task_func)
        def sync_wrapper(*args, **kwargs):
            # 同步循环重试
            start_time = time.time()
            for retry_count in range(max_retry_count):
                print(f"execute count {retry_count + 1}")
                use_time = time.time() - start_time
                if max_timeout and use_time > max_timeout:
                    # 超出最大超时时刻
                    print(f"execute timeout, use time {use_time}s, max timeout {max_timeout}")
                    return
                try:
                    task_ret = task_func(*args, **kwargs)
                    return task_ret
                except Exception as e:
                    print(f"fail {str(e)}")
                    time.sleep(time_interval)
        @functools.wraps(task_func)
        async def async_wrapper(*args, **kwargs):
            # 异步循环重试
            start_time = time.time()
            for retry_count in range(max_retry_count):
                print(f"execute count {retry_count + 1}")
                use_time = time.time() - start_time
                if max_timeout and use_time > max_timeout:
                    # 超出最大超时时刻
                    print(f"execute timeout, use time {use_time}s, max timeout {max_timeout}")
                    return
                try:
                    return await task_func(*args, **kwargs)
                except Exception as e:
                    print(f"fail {str(e)}")
                    await asyncio.sleep(time_interval)
        # 异步函数判断
        wrapper_func = async_wrapper if asyncio.iscoroutinefunction(task_func) else sync_wrapper
        return wrapper_func
    return _task_retry

留意时刻等待 await asyncio.sleep(time_interval) 会导致函数挂起，程序不会在这里等待，而是去事情循环loop中履行其他的现已安排妥当的使命，假如其他函数运行时刻太久了才切换回来，会导致时刻超时，换成 time.sleep()的话其实也没有用，假如函数内部还有异步函数履行仍是会切换出去，因而异步的时分感觉超时参数含义不大。

模仿测验下

@task_retry(max_retry_count=5, time_interval=2, max_timeout=5)
async def user_place_order_async():
    """异步函数重试事例"""
    a = 1 / 0
    print("user place order success")
    return {"code": 0, "msg": "ok"}
async def io_test():
    """模仿io堵塞"""
    print("io test start")
    time.sleep(3)
    print("io test end")
    return "io test end"
async def main():
    # 同步事例
    # ret = user_place_order()
    # print(f"user place order ret {ret}")
    # 异步事例
    # ret = await user_place_order_async()
    # print(f"user place order ret {ret}")
    # 并发异步
    order_ret, io_ret = await asyncio.gather(
        user_place_order_async(),
        io_test(),
    )
    print(f"io ret {io_ret}")
    print(f"user place order ret {order_ret}")
if __name__ == '__main__':
    asyncio.run(main())
 >>>out
execute count 1
fail division by zero
io test start
io test end
execute count 2
fail division by zero
execute count 3
execute timeout, use time 5.015768527984619s, max timeout 5
io ret io test end
user place order ret None

能够看出履行一遍后自动切换到了 io_test 中履行由于 io test 中的 time.sleep(3) 会导致整个线程堵塞，一定要等到io_test履行完后才会切换回去，然后再履行两遍就超时了，你或许会说都用异步的库，是的异步的库是能够加速，但我想表达便是这时分统计的耗时是整个程序的而不是独自一个函数的。我们能够在谈论区帮我想想有没有其他的方法，要么就不要用这个超时参数。

能够兼容异步函数、然后超时参数能够不装备，影响不大，O(∩_∩)O~

重试装修器-最终版

最终版便是运用抛反常的方法来完毕超越最大重试次数、最大超时，而不是直接回来None，然后再添加一个可装备捕获指定反常的参数，当产生特定反常的时分才重试。

import time
import asyncio
import functools
from typing import Type
class MaxRetryException(Exception):
    """最大重试次数反常"""
    pass
class MaxTimeoutException(Exception):
    """最大超时反常"""
    pass
def task_retry(
        max_retry_count: int = 5,
        time_interval: int = 2,
        max_timeout: int = None,
        catch_exc: Type[BaseException] = Exception
):
    """
    使命重试装修器
    Args:
        max_retry_count: 最大重试次数 默许 5 次
        time_interval: 每次重试距离 默许 2s
        max_timeout: 最大超时时刻，单位s 默许为 None,
        catch_exc: 指定捕获的反常类用于特定的反常重试 默许捕获 Exception
    """
    def _task_retry(task_func):
        @functools.wraps(task_func)
        def sync_wrapper(*args, **kwargs):
            # 函数循环重试
            start_time = time.time()
            for retry_count in range(max_retry_count):
                print(f"execute count {retry_count + 1}")
                use_time = time.time() - start_time
                if max_timeout and use_time > max_timeout:
                    # 超出最大超时时刻
                    raise MaxTimeoutException(f"execute timeout, use time {use_time}s, max timeout {max_timeout}")
                try:
                    task_ret = task_func(*args, **kwargs)
                    return task_ret
                except catch_exc as e:
                    print(f"fail {str(e)}")
                    time.sleep(time_interval)
            else:
                # 超越最大重试次数, 抛反常停止
                raise MaxRetryException(f"超越最大重试次数失利, max_retry_count {max_retry_count}")
        @functools.wraps(task_func)
        async def async_wrapper(*args, **kwargs):
            # 异步循环重试
            start_time = time.time()
            for retry_count in range(max_retry_count):
                print(f"execute count {retry_count + 1}")
                use_time = time.time() - start_time
                if max_timeout and use_time > max_timeout:
                    # 超出最大超时时刻
                    raise MaxTimeoutException(f"execute timeout, use time {use_time}s, max timeout {max_timeout}")
                try:
                    return await task_func(*args, **kwargs)
                except catch_exc as e:
                    print(f"fail {str(e)}")
                    await asyncio.sleep(time_interval)
            else:
                # 超越最大重试次数, 抛反常停止
                raise MaxRetryException(f"超越最大重试次数失利, max_retry_count {max_retry_count}")
        # 异步函数判断
        wrapper_func = async_wrapper if asyncio.iscoroutinefunction(task_func) else sync_wrapper
        return wrapper_func
    return _task_retry
@task_retry(max_retry_count=3, time_interval=1, catch_exc=ZeroDivisionError，max_timeout=5)
def user_place_order():
    a = 1 / 0
    print("user place order success")
    return {"code": 0, "msg": "ok"}
@task_retry(max_retry_count=5, time_interval=2, max_timeout=5)
async def user_place_order_async():
    """异步函数重试事例"""
    a = 1 / 0
    print("user place order success")
    return {"code": 0, "msg": "ok"}
async def io_test():
    """模仿io堵塞"""
    print("io test start")
    time.sleep(3)
    print("io test end")
    return "io test end"
async def main():
    # 同步事例
    try:
        ret = user_place_order()
        print(f"user place order ret {ret}")
    except MaxRetryException as e:
        # 超越最大重试次数处理
        print("MaxRetryException", e)
    except MaxTimeoutException as e:
        # 超越最大超时处理
        print("MaxTimeoutException", e)
    # 异步事例
    # ret = await user_place_order_async()
    # print(f"user place order ret {ret}")
    # 并发异步
    # order_ret, io_ret = await asyncio.gather(
    #     user_place_order_async(),
    #     io_test(),
    # )
    # print(f"io ret {io_ret}")
    # print(f"user place order ret {order_ret}")
if __name__ == '__main__':
    asyncio.run(main())

测验捕获指定反常

# 指定捕获除零过错，正常捕获重试
@task_retry(max_retry_count=3, time_interval=1, catch_exc=ZeroDivisionError)
def user_place_order():
    a = 1 / 0
    # a = []
    # b = a[0]
    print("user place order success")
    return {"code": 0, "msg": "ok"}
# out
execute count 1
fail division by zero
execute count 2
fail division by zero
execute count 3
fail division by zero
MaxRetryException 超越最大重试次数失利, max_retry_count 3
# 指定捕获除零过错，报索引越界过错，未正常捕获重试，直接退出
@task_retry(max_retry_count=3, time_interval=1, catch_exc=ZeroDivisionError)
def user_place_order():
    # a = 1 / 0
    a = []
    b = a[0]
    print("user place order success")
    return {"code": 0, "msg": "ok"}
# out
Traceback (most recent call last):
  File "G:/code/python/py-tools/decorator/base.py", line 184, in <module>
    asyncio.run(main())
  File "G:\softs\DevEnv\python-3.7.9\lib\asyncio\runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "G:\softs\DevEnv\python-3.7.9\lib\asyncio\base_events.py", line 587, in run_until_complete
    return future.result()
  File "G:/code/python/py-tools/decorator/base.py", line 161, in main
    ret = user_place_order()
  File "G:/code/python/py-tools/decorator/base.py", line 97, in sync_wrapper
    task_ret = task_func(*args, **kwargs)
  File "G:/code/python/py-tools/decorator/base.py", line 137, in user_place_order
    b = a[0]
IndexError: list index out of range
Process finished with exit code 1

用抛反常的方法推出更好知道是由于什么原因退出来进行对应的事务逻辑处理。乍一看兼容异步、同步函数导致这个装修器有点代码冗余，看看我们有没有更好的办法。

到这就完毕了，虽然是一个简略的装修器，但本文仍是引伸出很多知识点，望能够帮助到我们。