本文目录导读：
针对“脚本重试处理入库失败项”的需求，常见的场景包括：数据库写入失败、文件导入失败、API 调用失败等，下面给出一个通用的重试处理框架，并针对不同场景给出具体实现示例（Python）。
通用重试策略设计
| 策略 | 说明 | 适用场景 |
|---|---|---|
| 固定间隔重试 | 每次等待相同时间 | 网络波动 |
| 指数退避 | 等待时间逐步增加 | 服务端限流/负载高 |
| 带抖动的退避 | 增加随机延迟，避免惊群效应 | 分布式系统 |
| 最大次数限制 | 超过次数后记录失败或跳过 | 避免死循环 |
核心重试逻辑模板
import time
import random
from typing import Callable, Any
def retry_with_backoff(
func: Callable,
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
jitter: bool = True,
retryable_exceptions: tuple = (Exception,)
) -> Any:
"""
通用重试装饰器/函数
:param func: 要执行的任务函数
:param max_retries: 最大重试次数
:param base_delay: 基础等待秒数
:param max_delay: 最大等待秒数
:param jitter: 是否添加随机抖动
:param retryable_exceptions: 可重试的异常类型
:return: 函数执行结果
"""
for attempt in range(max_retries + 1):
try:
return func()
except retryable_exceptions as e:
if attempt >= max_retries:
raise # 超过最大重试次数，抛出最终异常
delay = min(base_delay * (2 ** attempt), max_delay)
if jitter:
delay = delay * (0.5 + random.random() * 0.5)
print(f"重试 #{attempt+1}/{max_retries}: 失败原因={e}, 等待 {delay:.2f}秒...")
time.sleep(delay)
实际入库重试场景示例
数据库写入重试
import pymysql
from sqlalchemy import create_engine
from sqlalchemy.exc import OperationalError, TimeoutError
def insert_with_retry(records, db_conn_str):
"""带重试的数据库批量插入"""
engine = create_engine(db_conn_str)
def do_insert():
with engine.connect() as conn:
conn.execute(
"INSERT INTO target_table (col1, col2) VALUES (%s, %s)",
records
)
return True
try:
retry_with_backoff(
do_insert,
max_retries=3,
retryable_exceptions=(OperationalError, TimeoutError)
)
return {"status": "success", "count": len(records)}
except Exception as e:
return {"status": "failed", "reason": str(e), "failed_records": records}
文件逐条入库并记录失败
def process_and_retry_import(file_path):
"""逐行读取文件并重试入库失败的行"""
engine = create_engine("mysql+pymysql://user:pass@host/db")
failed_rows = []
def insert_row(row_data):
with engine.connect() as conn:
conn.execute(
"INSERT INTO table (field1, field2) VALUES (%s, %s)",
(row_data['field1'], row_data['field2'])
)
with open(file_path, 'r') as f:
for line_no, line in enumerate(f, 1):
row_data = parse_line(line) # 你的解析函数
try:
retry_with_backoff(
lambda: insert_row(row_data),
max_retries=2,
retryable_exceptions=(OperationalError,)
)
except Exception as e:
failed_rows.append({
"line_no": line_no,
"data": row_data,
"error": str(e)
})
# 继续处理下一行，不中断整体流程
# 最终报告
if failed_rows:
save_failed_records(failed_rows) # 写入失败日志表
return {
"total": count_lines,
"success": count_lines - len(failed_rows),
"failed": len(failed_rows)
}
消息队列/API 回调重试
def process_kafka_message(message_value):
"""消费消息并重试入库"""
data = json.loads(message_value)
def save_to_db():
# 实际入库逻辑
r = requests.post("http://api.example.com/insert", json=data, timeout=5)
r.raise_for_status()
return r.json()
try:
result = retry_with_backoff(
save_to_db,
max_retries=5,
base_delay=0.5,
retryable_exceptions=(requests.Timeout, requests.ConnectionError)
)
return result
except Exception:
# 超过重试次数，放入死信队列或记录
send_to_dead_letter_queue(data)
raise
建议的失败处理策略
| 失败后动作 | 说明 | 示例代码 |
|---|---|---|
| 跳过并记录 | 跳过该条，记录日志 | logger.warning(...); continue |
| 放入重试队列 | 异步后补重试 | retry_queue.put(failed_item) |
| 死信队列 | 人工介入处理 | send_to_dlq(item) |
| 事务回滚+全量重试 | 批处理时整体回滚 | session.rollback(); time.sleep(...) |
最佳实践
-
区分可重试和不可重试错误
- 可重试：网络超时、连接断开、死锁、临时服务不可用
- 不可重试：数据格式错误、唯一键冲突、权限不足（应直接记录失败）
-
幂等性保证
- 重试操作必须是幂等的（重复执行结果相同）
- 例如使用
INSERT ... ON DUPLICATE KEY UPDATE或引入唯一业务ID
-
监控与告警
- 日志记录重试次数、失败原因、最终状态
- 超过阈值（如连续失败10条）触发告警
-
不要无限重试
- 总超时控制：
overall_timeout = 300秒后放弃 - 计数限制：
max_retries=5
- 总超时控制：
完整代码示例（带进度反馈）
import time
import random
from datetime import datetime
from typing import List, Dict
class RetryProcessor:
def __init__(self, max_retries=3, base_delay=1.0):
self.max_retries = max_retries
self.base_delay = base_delay
self.stats = {"total": 0, "success": 0, "failed": 0, "retries": 0}
def retry_operation(self, operation, record_id):
"""对单个操作进行重试"""
for attempt in range(self.max_retries + 1):
try:
result = operation()
self.stats["success"] += 1
return result
except Exception as e:
if attempt < self.max_retries:
delay = self.base_delay * (2 ** attempt)
delay += random.uniform(0, delay * 0.1)
self.stats["retries"] += 1
print(f"[{datetime.now()}] 重试 {record_id}: 第{attempt+1}次失败: {e}")
time.sleep(delay)
else:
self.stats["failed"] += 1
raise
def batch_process(self, records: List[Dict], save_func):
"""批量处理入库"""
failed_records = []
for record in records:
self.stats["total"] += 1
record_id = record.get("id", "unknown")
try:
self.retry_operation(lambda: save_func(record), record_id)
except Exception as e:
failed_records.append({"record": record, "error": str(e)})
return failed_records
# 使用示例
def main():
processor = RetryProcessor(max_retries=3, base_delay=1)
def save_to_db(record):
# 模拟可能失败的操作
if random.random() < 0.3:
raise Exception("数据库连接超时")
print(f" [成功] 入库记录: {record['id']}")
test_data = [{"id": i, "value": f"data_{i}"} for i in range(10)]
fails = processor.batch_process(test_data, save_to_db)
print(f"\n处理结果: {processor.stats}")
if fails:
print(f"失败记录数: {len(fails)}")
if __name__ == "__main__":
main()
- 核心：用
try-except+ 循环 + 退避策略 实现重试 - 关键：区分可重试错误、保证幂等性、设置重试上限
- 工程化：记录重试统计、失败日志、支持死信处理
根据你的实际场景（文件、数据库、API），直接套用上面的模板，调整 retryable_exceptions 和重试参数即可。
