本文目录导读：
我来系统地介绍构建稳定数据同步能力的Python方案,从基础到进阶。
基础同步框架
文件同步实现
import os
import hashlib
import json
from pathlib import Path
import shutil
from datetime import datetime
from typing import Dict, List, Optional
class FileSynchronizer:
"""文件同步器"""
def __init__(self, source: str, target: str, state_file: str = ".sync_state.json"):
self.source = Path(source)
self.target = Path(target)
self.state_file = self.target / state_file
self.state = self._load_state()
def _load_state(self) -> Dict:
"""加载同步状态"""
if self.state_file.exists():
with open(self.state_file, 'r') as f:
return json.load(f)
return {}
def _save_state(self):
"""保存同步状态"""
self.state_file.parent.mkdir(exist_ok=True)
with open(self.state_file, 'w') as f:
json.dump(self.state, f, indent=2)
def _file_hash(self, filepath: Path) -> str:
"""计算文件哈希"""
hasher = hashlib.sha256()
with open(filepath, 'rb') as f:
for chunk in iter(lambda: f.read(4096), b''):
hasher.update(chunk)
return hasher.hexdigest()
def _file_changed(self, filepath: Path) -> bool:
"""检查文件是否变更"""
rel_path = str(filepath.relative_to(self.source))
current_hash = self._file_hash(filepath)
return self.state.get(rel_path) != current_hash
def sync(self, dry_run: bool = False) -> List[Dict]:
"""执行同步"""
changes = []
for filepath in self.source.rglob('*'):
if filepath.is_file():
rel_path = str(filepath.relative_to(self.source))
target_path = self.target / rel_path
if self._file_changed(filepath):
change = {
'type': 'copy',
'source': str(filepath),
'target': str(target_path)
}
if not dry_run:
target_path.parent.mkdir(exist_ok=True)
shutil.copy2(filepath, target_path)
self.state[rel_path] = self._file_hash(filepath)
changes.append(change)
if not dry_run:
self._save_state()
return changes
数据库同步方案
增量同步实现
import sqlite3
import hashlib
import json
from datetime import datetime
from typing import List, Dict, Any
import threading
import time
class DatabaseSynchronizer:
"""数据库同步器"""
def __init__(self, source_conn, target_conn, sync_table: str):
self.source = source_conn
self.target = target_conn
self.sync_table = sync_table
self.lock = threading.Lock()
def _get_schema(self, conn) -> List[Dict]:
"""获取表结构"""
cursor = conn.execute(
f"PRAGMA table_info({self.sync_table})"
)
return cursor.fetchall()
def _generate_record_hash(self, record: tuple) -> str:
"""生成记录哈希"""
return hashlib.md5(str(record).encode()).hexdigest()
def _get_source_records(self, last_sync: str = None) -> List:
"""获取源端记录"""
if last_sync:
query = f"""
SELECT * FROM {self.sync_table}
WHERE updated_at > ?
ORDER BY updated_at
"""
return self.source.execute(query, (last_sync,)).fetchall()
else:
query = f"SELECT * FROM {self.sync_table}"
return self.source.execute(query).fetchall()
def incremental_sync(self, batch_size: int = 1000):
"""增量同步"""
schema = self._get_schema(self.source)
columns = [col[1] for col in schema]
placeholders = ','.join(['?' for _ in columns])
with self.lock:
last_sync = self._get_last_sync_time()
# 获取变更记录
records = self._get_source_records(last_sync)
# 批量处理
for i in range(0, len(records), batch_size):
batch = records[i:i + batch_size]
for record in batch:
record_hash = self._generate_record_hash(record)
# 检查目标端是否存在
cursor = self.target.execute(
f"SELECT hash FROM {self.sync_table}_sync WHERE id=?",
(record[0],)
)
existing = cursor.fetchone()
if existing and existing[0] == record_hash:
continue # 记录无变化
# 更新或插入
if existing:
update_query = f"""
UPDATE {self.sync_table}
SET {','.join([f'{col}=?' for col in columns])}
WHERE id=?
"""
self.target.execute(update_query, record + (record[0],))
else:
insert_query = f"""
INSERT INTO {self.sync_table} ({','.join(columns)})
VALUES ({placeholders})
"""
self.target.execute(insert_query, record)
# 更新同步状态
self.target.execute(
f"""
INSERT OR REPLACE INTO {self.sync_table}_sync (id, hash, synced_at)
VALUES (?, ?, ?)
""",
(record[0], record_hash, datetime.now())
)
self.target.commit()
self._update_sync_time(datetime.now())
def _get_last_sync_time(self) -> str:
"""获取上次同步时间"""
cursor = self.target.execute(
"SELECT last_sync FROM sync_metadata WHERE table_name=?",
(self.sync_table,)
)
result = cursor.fetchone()
return result[0] if result else None
def _update_sync_time(self, timestamp: datetime):
"""更新同步时间"""
self.target.execute(
"""
INSERT OR REPLACE INTO sync_metadata (table_name, last_sync)
VALUES (?, ?)
""",
(self.sync_table, timestamp)
)
self.target.commit()
分布式同步实现
基于消息队列的同步
import json
import redis
from typing import Callable, Optional
import pickle
import logging
class DistributedSynchronizer:
"""分布式同步器"""
def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis_client = redis.from_url(redis_url)
self.logger = logging.getLogger(__name__)
self.handlers = {}
def register_handler(self, event_type: str, handler: Callable):
"""注册事件处理器"""
self.handlers[event_type] = handler
def publish_event(self, event_type: str, data: dict):
"""发布同步事件"""
event = {
'type': event_type,
'data': data,
'timestamp': time.time()
}
# 序列化事件
serialized = pickle.dumps(event)
# 发布到Redis频道
self.redis_client.publish('sync_events', serialized)
self.logger.info(f"Published event: {event_type}")
def listen_events(self, timeout: int = 30):
"""监听同步事件"""
pubsub = self.redis_client.pubsub()
pubsub.subscribe('sync_events')
def process_event(message):
if message['type'] == 'message':
try:
event = pickle.loads(message['data'])
event_type = event['type']
if event_type in self.handlers:
self.handlers[event_type](event['data'])
self.logger.info(f"Processed event: {event_type}")
else:
self.logger.warning(f"No handler for event type: {event_type}")
except Exception as e:
self.logger.error(f"Failed to process event: {e}")
thread = threading.Thread(
target=self._listen_loop,
args=(pubsub, process_event, timeout)
)
thread.daemon = True
thread.start()
return thread
def _listen_loop(self, pubsub, callback, timeout):
"""监听循环"""
pubsub.subscribe('sync_events')
try:
for message in pubsub.listen():
callback(message)
except Exception as e:
self.logger.error(f"Listen loop error: {e}")
实时同步监控
健康检查与重试机制
from dataclasses import dataclass
from typing import Optional
import time
import random
@dataclass
class SyncMetrics:
"""同步指标"""
total_records: int = 0
success_count: int = 0
failure_count: int = 0
last_sync_time: Optional[datetime] = None
avg_sync_duration: float = 0.0
class SyncMonitor:
"""同步监控器"""
def __init__(self, max_retries: int = 3, retry_delay: int = 5):
self.max_retries = max_retries
self.retry_delay = retry_delay
self.metrics = SyncMetrics()
self.alerts = []
def execute_with_retry(self, sync_func: Callable, *args, **kwargs):
"""带重试的执行"""
for attempt in range(1, self.max_retries + 1):
try:
start_time = time.time()
result = sync_func(*args, **kwargs)
duration = time.time() - start_time
self.metrics.success_count += 1
self.metrics.last_sync_time = datetime.now()
self.metrics.avg_sync_duration = (
(self.metrics.avg_sync_duration * (self.metrics.success_count - 1) + duration)
/ self.metrics.success_count
)
return result
except Exception as e:
self.metrics.failure_count += 1
if attempt < self.max_retries:
wait_time = self.retry_delay * (2 ** (attempt - 1)) # 指数退避
wait_time += random.uniform(0, 1) # 添加抖动
self.logger.warning(
f"Sync attempt {attempt} failed, retrying in {wait_time:.2f}s"
)
time.sleep(wait_time)
else:
self._trigger_alert(f"Sync failed after {self.max_retries} attempts")
raise
def _trigger_alert(self, message: str):
"""触发告警"""
alert = {
'timestamp': datetime.now(),
'message': message,
'severity': 'high'
}
self.alerts.append(alert)
# 可以在这里集成告警系统
self.logger.error(f"ALERT: {message}")
def get_health_status(self) -> Dict:
"""获取健康状态"""
success_rate = (
self.metrics.success_count / (self.metrics.success_count + self.metrics.failure_count)
if self.metrics.total_records > 0 else 1.0
)
return {
'status': 'healthy' if success_rate > 0.95 else 'degraded',
'metrics': self.metrics.__dict__,
'alerts': self.alerts[-10:], # 最近10条告警
'success_rate': success_rate
}
最佳实践建议
冲突解决策略
class ConflictResolver:
"""冲突解决器"""
def __init__(self, strategy: str = 'last_write_wins'):
self.strategy = strategy
def resolve(self, local_record: Dict, remote_record: Dict) -> Dict:
"""解决冲突"""
if self.strategy == 'last_write_wins':
# 以最后修改为准
if local_record['updated_at'] > remote_record['updated_at']:
return local_record
return remote_record
elif self.strategy == 'manual':
# 手动解决冲突
conflict_id = f"{local_record['id']}_{datetime.now().timestamp()}"
return {
'conflict_id': conflict_id,
'local': local_record,
'remote': remote_record,
'status': 'unresolved'
}
elif self.strategy == 'merge':
# 智能合并
return self._merge_records(local_record, remote_record)
def _merge_records(self, local: Dict, remote: Dict) -> Dict:
"""智能合并记录"""
merged = {}
for key in set(list(local.keys()) + list(remote.keys())):
if key in local and key in remote:
if self._is_mergeable(key):
# 可合并的字段（如数组）
merged[key] = list(set(local[key] + remote[key]))
elif isinstance(local[key], dict) and isinstance(remote[key], dict):
# 嵌套字典合并
nested = ConflictResolver('merge')
merged[key] = nested.resolve(local[key], remote[key])
else:
# 取最新的非空值
merged[key] = local[key] if local[key] else remote[key]
elif key in local:
merged[key] = local[key]
else:
merged[key] = remote[key]
merged['merged_at'] = datetime.now().isoformat()
return merged
def _is_mergeable(self, key: str) -> bool:
"""检查字段是否可以合并"""
mergeable_fields = {'tags', 'permissions', 'extensions'}
return key in mergeable_fields
性能优化技巧
# 1. 批量处理
def batch_sync(records: List, batch_size: int = 500):
for i in range(0, len(records), batch_size):
batch = records[i:i + batch_size]
process_batch(batch)
time.sleep(0.1) # 控制速率
# 2. 并行处理
from concurrent.futures import ThreadPoolExecutor
def parallel_sync(records: List, num_workers: int = 4):
with ThreadPoolExecutor(max_workers=num_workers) as executor:
futures = [executor.submit(process_record, record) for record in records]
for future in futures:
future.result()
# 3. 增量同步
def get_changes_since(last_sync: datetime):
return """
SELECT * FROM table
WHERE updated_at > ?
ORDER BY updated_at
LIMIT 1000
"""
构建稳定数据同步的关键点：
- 状态追踪 - 维护同步状态避免重复工作
- 增量同步 - 只传输变更数据提高效率
- 错误处理 - 完善的重试和恢复机制
- 冲突解决 - 清晰的冲突处理策略
- 监控告警 - 实时掌握同步状态
- 性能优化 - 批量、并行、限速等技术
根据实际需求选择合适的同步策略,从小规模开始，逐步优化。
