本文目录导读：
我来介绍几种在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
- 状态持久化：使用文件或数据库保存进度
- 原子性操作：每个操作完成后保存状态
- 错误处理：捕获异常并保存当前进度
- 幂等性：确保重复执行不会产生副作用
- 监控机制：记录失败次数和原因
选择哪种方法取决于你的具体需求：
- 简单任务：使用状态文件
- 复杂任务：使用数据库
- 性能要求高：使用迭代器模式
