<p>可以通过在重试装饰器中引入异常类型过滤、日志记录和指数退避机制来实现更智能的重试策略。1. 通过设置exceptions参数指定仅对特定异常（如networkerror）进行重试，而对其他异常直接抛出；2. 使用logging模块在重试前记录警告日志，失败后记录错误日志，便于问题排查；3. 采用指数退避算法（如base_delay (2 * retries) + 随机抖动）动态增加重试间隔，避免请求堆积和服务器压力，从而有效防止无限重试并提升系统稳定性。最终方案结合了异常类型判断、日志输出和延迟递增策略，完整实现了高效、可控、可维护的函数重试机制。</p>
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中文网其它相关文章！
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移，垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是，许多工具可以让 Windows 保持平稳运行。
广告Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号
708
收藏
