在Python开发的实际场景中，网络请求是高频操作，但受网络不稳定、服务端负载过高、接口限流等因素影响，请求失败的情况十分常见。设计一套合理的网络重试机制，能够在请求失败时自动重新发起请求，有效降低偶发性故障对业务的影响，提升程序的健壮性。
网络重试机制的核心设计要点
设计重试机制不能盲目重试，需要遵循几个核心原则，避免带来新的问题：
- 明确重试触发条件：不是所有失败都需要重试，比如参数错误、权限不足这类客户端问题，重试多少次都不会成功，只有网络超时、服务端5xx错误这类临时性故障才适合重试。
- 控制重试次数：必须设置最大重试次数，避免无限重试导致程序卡死，或者大量无效请求占用资源。
- 设置重试间隔：连续快速重试容易给服务端造成压力，甚至触发限流，通常采用指数退避策略，每次重试的间隔逐步拉长。
- 异常类型过滤：只捕获需要重试的异常类型，其他异常直接抛出，避免掩盖真正的程序错误。
基于装饰器的轻量重试实现
如果只需要对单个网络请求函数添加重试逻辑，使用装饰器是最简洁的方式，不需要改动原有函数的核心逻辑。
import time
def retry(max_retries=3, delay=1, exceptions=(Exception,)):
"""
网络重试装饰器
:param max_retries: 最大重试次数
:param delay: 初始重试间隔（秒）
:param exceptions: 需要触发重试的异常类型元组
"""
def decorator(func):
def wrapper(*args, **kwargs):
last_exc = None
for i in range(max_retries + 1):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exc = e
# 最后一次重试失败，不再等待
if i == max_retries:
break
# 指数退避，每次间隔翻倍
wait_time = delay * (2 ** i)
time.sleep(wait_time)
# 所有重试都失败，抛出最后一次的异常
raise last_exc
return wrapper
return decorator
# 使用示例，模拟网络请求函数
import random
@retry(max_retries=3, delay=1, exceptions=(TimeoutError, ConnectionError))
def mock_network_request():
# 模拟30%概率请求失败
if random.random() < 0.3:
raise TimeoutError("请求超时")
return "请求成功"
# 调用函数
try:
result = mock_network_request()
print(result)
except Exception as e:
print(f"请求最终失败：{e}")
基于类的通用重试组件实现
如果需要在多个不同的网络请求场景中复用重试逻辑，或者需要更灵活的配置，可以封装一个重试类，支持自定义重试策略。
import time
class NetworkRetry:
def __init__(self, max_retries=3, base_delay=1, max_delay=10, retry_exceptions=(Exception,)):
"""
网络重试组件
:param max_retries: 最大重试次数
:param base_delay: 基础重试间隔（秒）
:param max_delay: 最大重试间隔上限（秒）
:param retry_exceptions: 触发重试的异常类型元组
"""
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.retry_exceptions = retry_exceptions
def execute(self, func, *args, **kwargs):
"""
执行带重试逻辑的函数
:param func: 需要执行的目标函数
:return: 函数执行结果
"""
last_exc = None
for i in range(self.max_retries + 1):
try:
return func(*args, **kwargs)
except self.retry_exceptions as e:
last_exc = e
if i == self.max_retries:
break
# 指数退避，同时限制最大间隔
wait_time = min(self.base_delay * (2 ** i), self.max_delay)
time.sleep(wait_time)
raise last_exc
# 使用示例
def another_mock_request():
import random
if random.random() < 0.4:
raise ConnectionError("连接失败")
return "另一个请求成功"
retry_tool = NetworkRetry(max_retries=2, base_delay=0.5, max_delay=5, retry_exceptions=(ConnectionError,))
try:
res = retry_tool.execute(another_mock_request)
print(res)
except Exception as e:
print(f"请求失败：{e}")
注意事项
在实际使用重试机制时，还需要注意几个问题：
- 如果请求是写操作（比如提交订单、修改数据），需要确认服务端是否支持幂等性，避免重试导致重复提交数据。
- 重试间隔不要设置过短，高并发场景下大量重试请求可能形成重试风暴，进一步加重服务端负担。
- 可以结合日志功能，记录每次重试的原因和次数，方便后续排查问题。
- 对于接口限流导致的失败，可以优先读取响应头中的重试时间，按照服务端要求的时间间隔重试，而不是固定使用指数退避。