一文教你實現Python重試裝飾器
更新時間：2023年02月13日 09:43:20 作者：憶想不到的暉
在寫業(yè)務代碼時候，有許多場景需要重試某塊業(yè)務邏輯，例如網絡請求、購物下單等，希望發(fā)生異常的時候多重試幾次。本文分享如何利用Python 的裝飾器來進行面向切面（AOP）的方式實現自動重試器，希望對大家有所幫助
一、前言
在寫業(yè)務代碼時候，有許多場景需要重試某塊業(yè)務邏輯，例如網絡請求、購物下單等，希望發(fā)生異常的時候多重試幾次。本文分享如何利用Python 的裝飾器來進行面向切面（AOP）的方式實現自動重試器。
二、簡單分析
一個重試裝飾器，最重要的就是發(fā)生意外異常處理失敗自動重試，有如下幾點需要注意
- 失敗不能一直重試，因為可能會出現死循環(huán)浪費資源，因此需要有 最大重試次數 或者 最大超時時間
- 不能重試太頻繁，因為太頻繁容易導致重試次數很快用完，卻沒有成功響應，需要有 重試時間間隔 來限制，有時可以加大成功概率，例如網絡請求時有一段時間是堵塞的，或者對方服務負載太高導致一段時間無法響應等。
簡單分析完，我們的重試裝飾器，就要支持可配置最大重試次數、最大超時時間、重試間隔，所以裝飾器就要設計成帶參數裝飾器。
三、代碼模擬實現
重試裝飾器-初版
分析完畢后，看看第一版的裝飾器
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):
# 函數循環(huán)重試
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 循環(huán) 來執(zhí)行指定重試次數，成功獲取結果就直接 return 返回，發(fā)生異常則睡眠配置重試間隔時間后繼續(xù)循環(huán)
寫個例子來模擬測試下看看效果
@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'}
沒有異常正常執(zhí)行，在函數中模擬一個異常來進行重試看看
@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 函數執(zhí)行了三遍，都發(fā)生了除零異常，最后超過最大執(zhí)行次數，返回了 None 值，我們可以在主邏輯中來判斷返回值是否為 None 來進行超過最大重試次數失敗的業(yè)務邏輯處理
ret = user_place_order()
print("user place order ret", ret)
if not ret:
print("user place order failed")
...重試裝飾器-改進版
現在只能配置 最大重試次數 沒有最大超時時間，有時候我們想不但有重試，還得在規(guī)定時間內完成，不想浪費太多試錯時間。所以增加一個 最大超時時間配置選項默認為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):
# 函數循環(huán)重試
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):
# 函數循環(huán)重試
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
發(fā)現發(fā)生異常的時候并沒有重試，為什么呢？其實在執(zhí)行 task_func() 它并沒有真正的執(zhí)行內部邏輯，而是返回一個 coroutine 協程對象，并不會報異常，所以再裝飾器中執(zhí)行一遍就成功就出來了，外面 ret = await user_place_order_async()， 后才真正的等待執(zhí)行，然后執(zhí)行函數內的邏輯再報異常就沒有捕獲到。我們可以打斷點驗證下
這樣裝飾器就不支持異步函數的重試，需要加強它，可以使用 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):
# 同步循環(huán)重試
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):
# 異步循環(huán)重試
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) 會導致函數掛起，程序不會在這里等待，而是去事件循環(huán)loop中執(zhí)行其他的已經就緒的任務，如果其他函數運行時間太久了才切換回來，會導致時間超時，換成 time.sleep()的話其實也沒有用，如果函數內部還有異步函數執(zhí)行還是會切換出去，因此異步的時候感覺超時參數意義不大。
模擬測試下
@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}")
# 并發(fā)異步
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
可以看出執(zhí)行一遍后自動切換到了 io_test 中執(zhí)行由于 io test 中的 time.sleep(3) 會導致整個線程阻塞，一定要等到io_test執(zhí)行完后才會切換回去，然后再執(zhí)行兩遍就超時了，你可能會說都用異步的庫，是的異步的庫是可以加速，但我想表達就是這時候統(tǒng)計的耗時是整個程序的而不是單獨一個函數的。大家可以在評論區(qū)幫我想想有沒有其他的方法，要么就不要用這個超時參數。
可以兼容異步函數、然后超時參數可以不配置，影響不大，O(∩_∩)O~
重試裝飾器-最終版
最終版就是利用拋異常的方式來結束超過最大重試次數、最大超時，而不是直接返回None，然后再添加一個可配置捕獲指定異常的參數，當發(fā)生特定異常的時候才重試。
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):
# 函數循環(huán)重試
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):
# 異步循環(huán)重試
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}")
# 并發(fā)異步
# 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
用拋異常的方式推出更好知道是因為什么原因退出來進行對應的業(yè)務邏輯處理。乍一看兼容異步、同步函數導致這個裝飾器有點代碼冗余，看看大家有沒有更好的辦法。
以上就是一文教你實現Python重試裝飾器的詳細內容，更多關于Python重試裝飾器的資料請關注腳本之家其它相關文章！
相關文章
這篇文章主要介紹了python requests.get帶heade方式，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編小編過來看看吧2020-05-05
這篇文章主要為大家詳細介紹了python實現數據寫入excel表格，具有一定的參考價值，感興趣的小伙伴們可以參考一下2018-03-03
作為一名合格的?Python?程序員，在工作中必然會用到二維碼相關操作，那如何快速的用?Python?實現呢？別著急，咱們這篇博客就為你解決2023-01-01
本文主要介紹了如何使用Django默認的Auth權限管理系統(tǒng)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2023-02-02
python完成FizzBuzzWhizz問題(拉勾網面試題)示例
這篇文章主要介紹了python完成FizzBuzzWhizz問題(拉勾網面試題)示例,需要的朋友可以參考下2014-05-05
今天小編就為大家分享一篇python 實現將list轉成字符串,中間用空格隔開，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-12-12
Pycharm中Python環(huán)境配置常見問題解析
這篇文章主要介紹了Pycharm中Python環(huán)境配置常見問題,結合圖文形式分析了Pycharm中Python環(huán)境配置模塊路徑問題、虛擬環(huán)境創(chuàng)建、配置遠程服務器、連接數據庫等常見問題與操作方法,需要的朋友可以參考下2020-01-01
在本篇文章里小編給大家整理的是關于Python作用領域等相關知識點，需要的朋友們可以參考下。2020-06-06
這篇文章主要介紹了Python 如何實現訪問者模式，文中講解非常細致，代碼幫助大家更好的理解和學習，感興趣的朋友可以了解下2020-07-07
Python 有很多黑魔法，為了不分你的心，今天只講 metaclass。對于 metaclass 這種特性，有兩種極端的觀點：下面小編將為大家詳細的介紹，剛興趣的小伙伴可以參考一下2021-09-09