詳解Python如何利用裝飾器實現(xiàn)重試機制

更新時間：2024年03月29日 14:53:45 作者：databook

重試機制在編程中是比較常見的場景,主要被用于處理那些可能由于臨時性故障或網(wǎng)絡波動等原因而失敗的操作,下面我們就來看看如何使用裝飾器來實現(xiàn)重試機制吧

1. 概要

關于Python的裝飾器，只是一個語法糖，原理也比較簡單，這里不在贅述。

關于為什么要用重試機制，

首先，它能顯著提高了系統(tǒng)的穩(wěn)定性和可靠性。

因為，在分布式系統(tǒng)、網(wǎng)絡通信或任何涉及外部資源調(diào)用的場景中，失敗和異常是難以避免的。

通過引入重試機制，系統(tǒng)能夠在遇到這些臨時性故障時自動恢復，減少因單次失敗導致的整體服務中斷。

其次，重試機制有助于提升用戶體驗。

對于用戶來說，如果系統(tǒng)因為一次網(wǎng)絡抖動或短暫的服務器不可用就拋出錯誤，那么用戶可能會感到不滿。

通過重試機制，系統(tǒng)可以在用戶幾乎無感知的情況下恢復服務，從而提升用戶體驗。

此外，重試機制還可以幫助系統(tǒng)更好地應對突發(fā)的高負載或資源緊張的情況。

當系統(tǒng)面臨大量請求或資源爭用時，某些操作可能會因為資源不足而失敗。

通過合理設置重試間隔和重試次數(shù)，系統(tǒng)可以平滑地處理這些突發(fā)情況，避免因為短暫的資源不足而導致服務崩潰。

2. 實現(xiàn)重試機制

下面是我目前在用的一個重試裝飾器：

from functools import wraps
from time import sleep


def retry(retries: int = 3, delay: float = 1):
    """
    函數(shù)執(zhí)行失敗時，重試

    :param retries: 最大重試的次數(shù)
    :param delay: 每次重試的間隔時間，單位 秒
    :return:
    """

    # 校驗重試的參數(shù)，參數(shù)值不正確時使用默認參數(shù)
    if retries < 1 or delay <= 0:
        retries = 3
        delay = 1

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 第一次正常執(zhí)行不算重試次數(shù)，所以retries+1
            for i in range(retries + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    # 檢查重試次數(shù)
                    if i == retries:
                        print(f"Error: {repr(e)}")
                        print(f'"{func.__name__}()" 執(zhí)行失敗，已重試{retries}次')
                        break
                    else:
                        print(
                            f"Error: {repr(e)}，{delay}秒后第[{i+1}/{retries}]次重試..."
                        )
                        sleep(delay)

        return wrapper

    return decorator

這個裝飾器有兩個參數(shù)，一個是重試次數(shù)(retries)，一個是每次重試的間隔(delay)。

代碼比較簡單，通過捕獲函數(shù)func的異常來重試，重試次數(shù)達到最大重試次數(shù)后退出。

3. 使用重試機制示例

使用上面裝飾器的示例：

from decorators import retry
import time


@retry(retries=2, delay=2)
def pay():
    now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    print(f"[{now}]: 開始調(diào)用支付接口")
    raise Exception("調(diào)用支付接口超時...")


@retry(retries=5, delay=1)
def third():
    now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    print(f"[{now}]: 開始調(diào)用第三方接口")
    raise Exception("調(diào)用第三方接口超時...")


if __name__ == "__main__":
    pay()  # 重試2次，每次間隔2秒
    third()  # 重試5次，每次間隔1秒

模擬一個支付接口，一個調(diào)用第三方的接口，分別看看重試的效果。