脚本怎样重试处理入库失败项

本文目录导读：

1. 通用重试策略设计
2. 核心重试逻辑模板
3. 实际入库重试场景示例
4. 建议的失败处理策略
5. 最佳实践
6. 完整代码示例（带进度反馈）

针对“脚本重试处理入库失败项”的需求，常见的场景包括：数据库写入失败、文件导入失败、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

建议的失败处理策略

最佳实践

1. 区分可重试和不可重试错误

   • 可重试：网络超时、连接断开、死锁、临时服务不可用
   • 不可重试：数据格式错误、唯一键冲突、权限不足（应直接记录失败）

2. 幂等性保证

   • 重试操作必须是幂等的（重复执行结果相同）
   • 例如使用 INSERT ... ON DUPLICATE KEY UPDATE 或引入唯一业务ID

3. 监控与告警

   • 日志记录重试次数、失败原因、最终状态
   • 超过阈值（如连续失败10条）触发告警

4. 不要无限重试

   • 总超时控制：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 和重试参数即可。