在Python的for循环处理批量任务时，部分元素执行过程中可能出现临时异常，比如网络波动导致的请求失败、资源暂时不可用等问题。如果直接跳过失败元素，可能会导致数据缺失或任务不完整，因此需要在不跳过当前元素的前提下实现失败重试机制。
基础重试逻辑实现
核心思路是在处理单个元素时，使用内层循环控制重试次数，只有当重试次数耗尽且仍然失败时，才结束当前元素的处理，继续下一个元素。我们可以通过for循环嵌套或者while循环配合重试计数器来实现。
下面是一个简单的示例，模拟处理元素时可能出现的异常，最多重试3次，每次重试间隔1秒：
import time
def process_element(item):
# 模拟处理元素，随机抛出异常
import random
if random.random() < 0.5:
raise ValueError(f"处理元素 {item} 失败")
print(f"元素 {item} 处理成功")
def for_loop_with_retry(elements, max_retries=3, retry_interval=1):
for item in elements:
retry_count = 0
success = False
# 内层循环控制当前元素的重试
while retry_count < max_retries:
try:
process_element(item)
success = True
break # 处理成功，跳出重试循环
except Exception as e:
retry_count += 1
print(f"元素 {item} 处理失败，第 {retry_count} 次重试，错误信息：{e}")
if retry_count < max_retries:
time.sleep(retry_interval) # 重试间隔
if not success:
print(f"元素 {item} 重试 {max_retries} 次后仍失败，跳过该元素继续后续处理")
print("所有元素处理完成")
# 测试代码
test_elements = [1, 2, 3, 4, 5]
for_loop_with_retry(test_elements)
带自定义异常过滤的重试实现
实际业务中，我们可能只希望对特定类型的异常进行重试，比如网络异常可以重试，但是参数错误导致的异常不需要重试。这时候可以在重试逻辑中增加异常类型的判断。
import time
import requests
def request_api(url):
# 模拟发送请求，网络异常重试，参数错误不重试
response = requests.get(url, timeout=3)
return response.status_code
def for_loop_with_custom_retry(urls, max_retries=3, retry_interval=1):
for url in urls:
retry_count = 0
success = False
while retry_count < max_retries:
try:
status = request_api(url)
print(f"请求 {url} 成功，状态码：{status}")
success = True
break
except requests.exceptions.Timeout as e:
# 超时异常，进行重试
retry_count += 1
print(f"请求 {url} 超时，第 {retry_count} 次重试")
if retry_count < max_retries:
time.sleep(retry_interval)
except requests.exceptions.HTTPError as e:
# HTTP错误，根据状态码判断是否需要重试
if 500 <= e.response.status_code < 600:
retry_count += 1
print(f"请求 {url} 返回5xx错误，第 {retry_count} 次重试")
if retry_count < max_retries:
time.sleep(retry_interval)
else:
# 4xx错误不重试
print(f"请求 {url} 返回4xx错误，不重试，错误信息：{e}")
break
except Exception as e:
# 其他未知异常不重试
print(f"请求 {url} 出现未知异常，不重试，错误信息：{e}")
break
if not success:
print(f"请求 {url} 重试 {max_retries} 次后仍失败或无需重试")
print("所有请求处理完成")
# 测试代码，使用ipipp.com模拟测试地址
test_urls = [
"http://ipipp.com/api/test1",
"http://ipipp.com/api/test2",
"http://ipipp.com/api/test3"
]
# 注意：实际运行时需要安装requests库，且测试地址需要可访问
# for_loop_with_custom_retry(test_urls)
重试机制的注意事项
- 重试次数不宜设置过高，避免任务执行时间过长，一般建议设置3-5次即可。
- 重试间隔可以根据业务场景调整，如果是网络请求类任务，建议间隔逐步增加，避免对服务端造成过大压力。
- 需要明确哪些异常需要重试，哪些异常不需要重试，避免无效重试。
- 如果重试次数耗尽仍然失败，需要根据业务需求决定是记录日志、抛出异常还是跳过当前元素，保证整体任务的稳定性。
通用重试装饰器封装
如果多个处理逻辑都需要重试功能，可以把重试逻辑封装成装饰器，减少重复代码。下面是一个通用的重试装饰器示例：
import time
from functools import wraps
def retry_decorator(max_retries=3, retry_interval=1, exceptions=(Exception,)):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
retry_count = 0
while retry_count < max_retries:
try:
return func(*args, **kwargs)
except exceptions as e:
retry_count += 1
if retry_count < max_retries:
print(f"函数 {func.__name__} 执行失败，第 {retry_count} 次重试，错误：{e}")
time.sleep(retry_interval)
else:
print(f"函数 {func.__name__} 重试 {max_retries} 次后仍失败，错误：{e}")
raise # 重试耗尽后抛出异常
return None
return wrapper
return decorator
# 使用装饰器实现重试
@retry_decorator(max_retries=3, retry_interval=1, exceptions=(ValueError,))
def process_single_item(item):
import random
if random.random() < 0.5:
raise ValueError(f"处理 {item} 失败")
print(f"处理 {item} 成功")
return item
def for_loop_with_decorator(elements):
for item in elements:
try:
process_single_item(item)
except Exception as e:
print(f"元素 {item} 处理最终失败，错误信息：{e}")
print("所有元素处理完成")
# 测试代码
test_items = [10, 20, 30, 40]
for_loop_with_decorator(test_items)