Python脚本如何保证多次同步结果一致

wen python案例 2026-06-25 291

本文目录导读：

文件/数据同步的幂等性
数据库同步的一致性
API数据同步的一致性
分布式锁确保一致性
最佳实践总结

保证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