可以通过在重试装饰器中引入异常类型过滤、日志记录和指数退避机制来实现更智能的重试策略。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)是一个随机抖动，用于避免多个重试请求同时到达服务器。
