本文目录导读：
保证Python脚本多次同步结果一致,核心问题是确定性和幂等性，以下是几种常见的场景和解决方案：
文件/数据同步的幂等性
使用哈希校验
import hashlib
import os
def get_file_hash(filepath):
"""计算文件哈希值"""
sha256_hash = hashlib.sha256()
with open(filepath, "rb") as f:
for byte_block in iter(lambda: f.read(4096), b""):
sha256_hash.update(byte_block)
return sha256_hash.hexdigest()
def sync_file(src, dst):
"""幂等文件同步"""
# 检查目标是否存在且内容相同
if os.path.exists(dst):
if get_file_hash(src) == get_file_hash(dst):
print(f"文件已同步: {dst}")
return True
# 执行同步
with open(src, 'rb') as f_src:
with open(dst, 'wb') as f_dst:
f_dst.write(f_src.read())
# 验证同步结果
assert get_file_hash(src) == get_file_hash(dst)
print(f"同步完成: {dst}")
return True
使用版本号或时间戳
import json
import os
class SyncTracker:
def __init__(self, state_file="sync_state.json"):
self.state_file = state_file
self.state = self.load_state()
def load_state(self):
"""加载同步状态"""
if os.path.exists(self.state_file):
with open(self.state_file, 'r') as f:
return json.load(f)
return {}
def save_state(self):
"""保存同步状态"""
with open(self.state_file, 'w') as f:
json.dump(self.state, f, indent=2)
def need_sync(self, item_id, item_version):
"""检查是否需要同步"""
if item_id not in self.state:
return True
return self.state[item_id] != item_version
def mark_synced(self, item_id, item_version):
"""标记已同步"""
self.state[item_id] = item_version
self.save_state()
数据库同步的一致性
使用事务和唯一约束
import sqlite3
class DatabaseSync:
def __init__(self, db_path):
self.db_path = db_path
self.init_db()
def init_db(self):
"""初始化数据库结构"""
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS sync_items (
id TEXT PRIMARY KEY,
data TEXT NOT NULL,
version INTEGER NOT NULL,
sync_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
def upsert_item(self, item_id, data, version):
"""幂等插入或更新"""
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
INSERT OR REPLACE INTO sync_items (id, data, version)
VALUES (?, ?, ?)
WHERE id NOT IN (
SELECT id FROM sync_items WHERE version >= ?
)
""", (item_id, data, version, version))
def get_sync_status(self):
"""获取同步状态"""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.execute("""
SELECT id, version FROM sync_items
""")
return dict(cursor.fetchall())
API数据同步的一致性
使用轮询和去重
import requests
import json
from datetime import datetime
class APISyncClient:
def __init__(self, api_url, last_sync_file="last_sync.json"):
self.api_url = api_url
self.last_sync_file = last_sync_file
self.processed_ids = set()
def load_last_sync(self):
"""加载上次同步状态"""
try:
with open(self.last_sync_file, 'r') as f:
return json.load(f)
except FileNotFoundError:
return {"last_timestamp": None, "processed_ids": []}
def save_last_sync(self, timestamp, processed_ids):
"""保存同步状态"""
state = {
"last_timestamp": timestamp,
"processed_ids": list(processed_ids)
}
with open(self.last_sync_file, 'w') as f:
json.dump(state, f)
def sync_data(self):
"""同步API数据"""
state = self.load_last_sync()
self.processed_ids = set(state.get("processed_ids", []))
last_timestamp = state.get("last_timestamp")
if last_timestamp:
params = {"since": last_timestamp}
else:
params = {}
response = requests.get(self.api_url, params=params)
data = response.json()
new_ids = set()
for item in data.get("items", []):
item_id = item["id"]
# 去重检查
if item_id in self.processed_ids:
continue
# 处理数据
self.process_item(item)
new_ids.add(item_id)
# 更新状态
current_timestamp = datetime.now().isoformat()
self.processed_ids.update(new_ids)
self.save_last_sync(current_timestamp, self.processed_ids)
def process_item(self, item):
"""处理单个数据项"""
# 实现数据处理逻辑
print(f"Processing: {item}")
分布式锁确保一致性
import threading
import time
from functools import wraps
class DistributedLock:
def __init__(self):
self.lock = threading.Lock()
self.operation_log = []
def synchronized(self, func):
@wraps(func)
def wrapper(*args, **kwargs):
with self.lock:
operation_id = f"{func.__name__}_{time.time()}"
# 检查是否已执行过
if operation_id in self.operation_log:
print(f"操作 {operation_id} 已执行，跳过")
return None
result = func(*args, **kwargs)
self.operation_log.append(operation_id)
return result
return wrapper
# 使用示例
lock = DistributedLock()
@lock.synchronized
def critical_sync_operation():
"""关键同步操作"""
time.sleep(1) # 模拟耗时操作
return "Operation completed"
# 多次调用，只有第一次会实际执行
critical_sync_operation()
critical_sync_operation() # 跳过
最佳实践总结
状态管理
class SyncManager:
"""统一的同步管理器"""
def __init__(self):
self.sync_state = {}
self.lock = threading.Lock()
def try_sync(self, item_id, sync_func):
"""尝试同步，返回是否成功执行"""
with self.lock:
if item_id in self.sync_state:
print(f"跳过已同步项: {item_id}")
return False
try:
sync_func()
self.sync_state[item_id] = True
return True
except Exception as e:
print(f"同步失败: {e}")
return False
关键原则
- 使用唯一标识符：每个同步项应有唯一ID
- 保存同步状态：记录已同步的ID和时间戳
- 实现幂等操作：相同操作多次执行结果相同
- 添加版本控制：避免旧数据覆盖新数据
- 使用事务：确保操作的原子性
- 校验机制：同步后验证数据一致性
这些方法可以组合使用,根据具体场景选择合适的策略，关键在于追踪状态和避免重复操作。
