Python脚本如何恢复节点后继续同步任务

wen python案例 2026-06-25 243

本文目录导读：

使用状态文件保存进度
使用数据库记录任务状态
使用迭代器模式
完整的恢复示例
使用装饰器实现自动恢复

我来介绍几种在Python中恢复节点后继续同步任务的策略：

使用状态文件保存进度

import json
import os
from pathlib import Path
class SyncWithCheckpoint:
    def __init__(self, checkpoint_file="sync_checkpoint.json"):
        self.checkpoint_file = checkpoint_file
        self.checkpoint = self.load_checkpoint()
    def load_checkpoint(self):
        """加载检查点"""
        if os.path.exists(self.checkpoint_file):
            with open(self.checkpoint_file, 'r') as f:
                return json.load(f)
        return {'last_processed_id': 0, 'processed_count': 0}
    def save_checkpoint(self):
        """保存检查点"""
        with open(self.checkpoint_file, 'w') as f:
            json.dump(self.checkpoint, f, indent=2)
    def sync_data(self, data_source):
        """同步数据，支持断点续传"""
        last_id = self.checkpoint['last_processed_id']
        for item in data_source[last_id:]:
            try:
                # 处理数据
                process_item(item)
                # 更新检查点
                self.checkpoint['last_processed_id'] = item['id']
                self.checkpoint['processed_count'] += 1
                self.save_checkpoint()
            except Exception as e:
                print(f"处理 {item['id']} 时出错: {e}")
                # 保存当前进度，下次从失败处继续
                self.save_checkpoint()
                raise

使用数据库记录任务状态

import sqlite3
from datetime import datetime
class TaskTracker:
    def __init__(self, db_path="sync_tasks.db"):
        self.conn = sqlite3.connect(db_path)
        self.create_table()
    def create_table(self):
        cursor = self.conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS tasks (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                task_name TEXT NOT NULL,
                status TEXT DEFAULT 'pending',
                progress INTEGER DEFAULT 0,
                last_updated TIMESTAMP,
                error_msg TEXT
            )
        ''')
        self.conn.commit()
    def resume_unfinished_tasks(self):
        """恢复未完成的任务"""
        cursor = self.conn.cursor()
        cursor.execute("SELECT * FROM tasks WHERE status IN ('pending', 'failed')")
        unfinished_tasks = cursor.fetchall()
        for task in unfinished_tasks:
            print(f"恢复任务: {task[1]}, 当前进度: {task[3]}%")
            self.process_task(task[1], task[3])
    def update_task_progress(self, task_name, progress, status='processing'):
        cursor = self.conn.cursor()
        cursor.execute("""
            UPDATE tasks 
            SET progress = ?, status = ?, last_updated = ?
            WHERE task_name = ?
        """, (progress, status, datetime.now(), task_name))
        self.conn.commit()

使用迭代器模式

class ResilientIterator:
    """支持断点续传的迭代器"""
    def __init__(self, data, checkpoint_file="iterator_checkpoint.txt"):
        self.data = data
        self.checkpoint_file = checkpoint_file
        self.current_index = self.load_progress()
    def load_progress(self):
        try:
            with open(self.checkpoint_file, 'r') as f:
                return int(f.read().strip())
        except (FileNotFoundError, ValueError):
            return 0
    def save_progress(self):
        with open(self.checkpoint_file, 'w') as f:
            f.write(str(self.current_index))
    def __iter__(self):
        return self
    def __next__(self):
        if self.current_index >= len(self.data):
            raise StopIteration
        item = self.data[self.current_index]
        self.current_index += 1
        self.save_progress()
        return item
# 使用示例
def sync_with_resilience(items):
    """使用恢复机制的同步函数"""
    iterator = ResilientIterator(items)
    for item in iterator:
        try:
            # 执行同步操作
            sync_item(item)
        except Exception as e:
            print(f"同步 {item} 失败: {e}")
            # 保存当前进度后退出
            iterator.save_progress()
            raise

完整的恢复示例

import time
import random
import json
from pathlib import Path
class DataSyncManager:
    def __init__(self, state_file="sync_state.json"):
        self.state_file = state_file
        self.state = self.load_state()
    def load_state(self):
        """加载同步状态"""
        if Path(self.state_file).exists():
            with open(self.state_file, 'r') as f:
                return json.load(f)
        return {
            'batch_index': 0,
            'item_index': 0,
            'started_at': None,
            'completed_items': []
        }
    def save_state(self):
        """保存同步状态"""
        with open(self.state_file, 'w') as f:
            json.dump(self.state, f, indent=2)
    def sync_data(self, data_batches):
        """执行同步，支持断点续传"""
        print(f"从批次 {self.state['batch_index']}, 项目 {self.state['item_index']} 开始同步")
        for batch_idx, batch in enumerate(data_batches[self.state['batch_index']:], 
                                          start=self.state['batch_index']):
            start_item = self.state['item_index'] if batch_idx == self.state['batch_index'] else 0
            for item_idx, item in enumerate(batch[start_item:], start=start_item):
                try:
                    # 模拟同步操作
                    print(f"同步批次 {batch_idx}, 项目 {item_idx}: {item}")
                    time.sleep(random.uniform(0.1, 0.5))
                    # 模拟随机失败
                    if random.random() < 0.2:  # 20% 概率失败
                        raise ConnectionError(f"节点 {item} 连接失败")
                    # 记录完成的项目
                    self.state['completed_items'].append({
                        'batch': batch_idx,
                        'item': item_idx,
                        'data': item,
                        'completed_at': time.time()
                    })
                    # 每处理10个项目保存一次状态
                    if len(self.state['completed_items']) % 10 == 0:
                        self.state['item_index'] = item_idx + 1
                        self.state['batch_index'] = batch_idx
                        self.save_state()
                except Exception as e:
                    print(f"同步失败 (批次 {batch_idx}, 项目 {item_idx}): {e}")
                    self.state['item_index'] = item_idx
                    self.state['batch_index'] = batch_idx
                    self.save_state()
                    return False  # 返回False表示需要重试
            # 批次完成，重置项目索引
            self.state['item_index'] = 0
            self.state['batch_index'] = batch_idx + 1
            self.save_state()
        # 所有同步完成
        self.state['completed'] = True
        self.save_state()
        print("同步完成!")
        return True
# 使用示例
def main():
    # 准备数据
    data = [
        [f"data_{i}_{j}" for j in range(10)]
        for i in range(5)
    ]
    manager = DataSyncManager()
    # 同步数据，失败会自动恢复
    while True:
        success = manager.sync_data(data)
        if success:
            break
        print("等待重试...")
        time.sleep(5)
if __name__ == "__main__":
    main()

使用装饰器实现自动恢复

import functools
import time
def auto_resume(max_retries=3, delay=5):
    """自动恢复装饰器"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except ConnectionError as e:
                    last_error = e
                    print(f"连接失败 (尝试 {attempt + 1}/{max_retries}): 等待 {delay} 秒重试")
                    time.sleep(delay)
            raise last_error
        return wrapper
    return decorator
@auto_resume(max_retries=3, delay=2)
def sync_node_data(node_id, data):
    """同步单个节点的数据"""
    # 同步逻辑
    pass