Python重试装饰器使用技巧

小伙伴们有没有觉得学习文章很有意思？有意思就对了！今天就给大家带来《Python函数重试装饰器入门技巧》，以下内容将会涉及到，若是在学习中对其中部分知识点有疑问，或许看了本文就能帮到你！

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

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重试装饰器使用技巧》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注golang学习网公众号，带你了解更多关于Python,日志记录,异常处理,指数退避,函数重试装饰器的知识点！