Python函数怎样用装饰器实现函数重试机制 Python函数重试装饰器的入门应用技巧​

可以通过在重试装饰器中引入异常类型过滤、日志记录和指数退避机制来实现更智能的重试策略。1. 通过设置exceptions参数指定仅对特定异常（如networkerror）进行重试，而对其他异常直接抛出；2. 使用logging模块在重试前记录警告日志，失败后记录错误日志，便于问题排查；3. 采用指数退避算法（如base_delay (2 * retries) + 随机抖动）动态增加重试间隔，避免请求堆积和服务器压力，从而有效防止无限重试并提升系统稳定性。最终方案结合了异常类型判断、日志输出和延迟递增策略，完整实现了高效、可控、可维护的函数重试机制。

Python函数重试装饰器通过在函数执行失败时自动重新调用该函数，提高程序的稳定性和容错能力。

解决方案：

函数重试装饰器的核心思想是：当被装饰的函数抛出异常时，装饰器会捕获这个异常，并根据设定的重试次数和延迟时间，自动重新执行该函数。如果超过最大重试次数，仍然失败，则抛出异常。

立即学习“Python免费学习笔记（深入）”；

一个简单的Python函数重试装饰器实现如下：

import time
import functools

def retry(max_retries=3, delay=1):
    """
    重试装饰器。

    Args:
        max_retries: 最大重试次数，默认为3。
        delay: 重试间隔时间，默认为1秒。
    """
    def decorator_retry(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            retries = 0
            while retries < max_retries:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    print(f"函数 {func.__name__} 失败，重试中... (第 {retries + 1} 次)")
                    retries += 1
                    time.sleep(delay)
            print(f"函数 {func.__name__} 达到最大重试次数，仍然失败。")
            raise  # 重新抛出异常
        return wrapper
    return decorator_retry

# 示例
@retry(max_retries=5, delay=2)
def unreliable_function():
    """
    一个可能失败的函数。
    """
    import random
    if random.randint(0, 9) < 7:  # 70%的概率失败
        raise Exception("模拟函数执行失败")
    print("函数执行成功！")
    return "成功"

if __name__ == '__main__':
    try:
        result = unreliable_function()
        print(f"函数返回结果: {result}")
    except Exception as e:
        print(f"最终失败: {e}")

这个例子中，

retry

是一个装饰器工厂函数，它接受

max_retries

和

delay

作为参数，用于配置重试次数和间隔时间。

decorator_retry

是真正的装饰器，它接受一个函数

func

作为参数，并返回一个新的函数

wrapper

。

wrapper

函数负责执行

func

，并在发生异常时进行重试。

functools.wraps(func)

用于保留原始函数的元信息，例如函数名和文档字符串。

如何根据不同的异常类型进行不同的重试策略？

有时候，我们希望针对不同的异常类型采取不同的重试策略。例如，对于网络连接错误，我们可能希望立即重试，而对于参数错误，我们则不希望重试。

可以修改装饰器，使其接受一个

exceptions

参数，用于指定需要重试的异常类型：

import time
import functools

def retry(exceptions=(Exception,), max_retries=3, delay=1):
    """
    重试装饰器，可以指定需要重试的异常类型。

    Args:
        exceptions: 需要重试的异常类型，默认为Exception。
        max_retries: 最大重试次数，默认为3。
        delay: 重试间隔时间，默认为1秒。
    """
    def decorator_retry(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            retries = 0
            while retries < max_retries:
                try:
                    return func(*args, **kwargs)
                except exceptions as e: # 只捕获指定的异常类型
                    print(f"函数 {func.__name__} 遇到 {type(e).__name__} 异常，重试中... (第 {retries + 1} 次)")
                    retries += 1
                    time.sleep(delay)
                except Exception as e:
                    # 捕获其他异常，直接抛出
                    print(f"函数 {func.__name__} 遇到 {type(e).__name__} 异常，不进行重试。")
                    raise
            print(f"函数 {func.__name__} 达到最大重试次数，仍然失败。")
            raise  # 重新抛出异常
        return wrapper
    return decorator_retry


# 示例
class NetworkError(Exception):
    pass

@retry(exceptions=(NetworkError,), max_retries=5, delay=2)
def unreliable_network_function():
    """
    一个可能因为网络问题失败的函数。
    """
    import random
    if random.randint(0, 9) < 7:
        raise NetworkError("模拟网络连接失败")
    print("网络函数执行成功！")
    return "成功"

if __name__ == '__main__':
    try:
        result = unreliable_network_function()
        print(f"函数返回结果: {result}")
    except Exception as e:
        print(f"最终失败: {e}")

这个例子中，

retry

装饰器现在接受一个

exceptions

参数，用于指定需要重试的异常类型。只有当函数抛出的异常类型在

exceptions

中时，才会进行重试。

如何在重试过程中记录日志？

记录日志可以帮助我们更好地了解函数执行的情况，并在出现问题时进行排查。可以在装饰器中加入日志记录功能：

import time
import functools
import logging

logging.basicConfig(level=logging.INFO)  # 配置日志

def retry(exceptions=(Exception,), max_retries=3, delay=1):
    """
    重试装饰器，可以指定需要重试的异常类型，并记录日志。

    Args:
        exceptions: 需要重试的异常类型，默认为Exception。
        max_retries: 最大重试次数，默认为3。
        delay: 重试间隔时间，默认为1秒。
    """
    def decorator_retry(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            retries = 0
            while retries < max_retries:
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    logging.warning(f"函数 {func.__name__} 遇到 {type(e).__name__} 异常，重试中... (第 {retries + 1} 次)")
                    retries += 1
                    time.sleep(delay)
                except Exception as e:
                    logging.error(f"函数 {func.__name__} 遇到 {type(e).__name__} 异常，不进行重试。")
                    raise
            logging.error(f"函数 {func.__name__} 达到最大重试次数，仍然失败。")
            raise  # 重新抛出异常
        return wrapper
    return decorator_retry

# 示例
@retry(max_retries=3, delay=1)
def another_unreliable_function():
    import random
    if random.randint(0, 9) < 5:
        raise ValueError("模拟值错误")
    print("函数执行成功！")
    return "成功"

if __name__ == '__main__':
    try:
        result = another_unreliable_function()
        print(f"函数返回结果: {result}")
    except Exception as e:
        print(f"最终失败: {e}")

这个例子中，我们使用了

logging

模块来记录日志。在每次重试之前，我们使用

logging.warning

记录一条警告信息。如果达到最大重试次数仍然失败，我们使用

logging.error

记录一条错误信息。

如何避免无限重试？

如果函数总是抛出异常，那么重试装饰器可能会陷入无限循环。为了避免这种情况，我们需要设置一个最大重试次数。在上面的例子中，我们已经使用了

max_retries

参数来限制重试次数。但是，如果

max_retries

设置得过大，仍然可能导致程序长时间阻塞。

一种更健壮的方法是使用指数退避算法来调整重试间隔时间。指数退避算法会随着重试次数的增加，逐渐增加重试间隔时间。这样可以避免在高并发情况下，大量重试请求同时到达服务器，导致服务器崩溃。

一个使用指数退避算法的重试装饰器实现如下：

import time
import functools
import random

def retry_with_backoff(exceptions=(Exception,), max_retries=5, base_delay=1):
    """
    使用指数退避算法的重试装饰器。

    Args:
        exceptions: 需要重试的异常类型，默认为Exception。
        max_retries: 最大重试次数，默认为5。
        base_delay: 基础延迟时间，默认为1秒。
    """
    def decorator_retry(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            retries = 0
            while retries < max_retries:
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    delay = base_delay * (2 ** retries) + random.uniform(0, 1)  # 指数退避 + 随机抖动
                    print(f"函数 {func.__name__} 遇到 {type(e).__name__} 异常，重试中... (第 {retries + 1} 次)，延迟 {delay:.2f} 秒")
                    retries += 1
                    time.sleep(delay)
                except Exception as e:
                    print(f"函数 {func.__name__} 遇到 {type(e).__name__} 异常，不进行重试。")
                    raise
            print(f"函数 {func.__name__} 达到最大重试次数，仍然失败。")
            raise  # 重新抛出异常
        return wrapper
    return decorator_retry

# 示例
@retry_with_backoff(max_retries=4, base_delay=0.5)
def yet_another_unreliable_function():
    import random
    if random.randint(0, 9) < 6:
        raise IOError("模拟IO错误")
    print("函数执行成功！")
    return "成功"

if __name__ == '__main__':
    try:
        result = yet_another_unreliable_function()
        print(f"函数返回结果: {result}")
    except Exception as e:
        print(f"最终失败: {e}")

这个例子中，我们使用

base_delay * (2 ** retries) + random.uniform(0, 1)

来计算重试间隔时间。其中，

base_delay

是基础延迟时间，

2 ** retries

是指数退避因子，

random.uniform(0, 1)

是一个随机抖动，用于避免多个重试请求同时到达服务器。

以上就是Python函数怎样用装饰器实现函数重试机制 Python函数重试装饰器的入门应用技巧​的详细内容，更多请关注php中文网其它相关文章！