本文目录导读：
我来为你介绍几种按需同步业务模块数据的Python实现方案：
基于配置驱动的同步方案
import json
import time
from typing import Dict, List, Optional
from datetime import datetime
import logging
class BusinessDataSync:
"""业务数据同步器"""
def __init__(self, config_file: str = 'sync_config.json'):
self.logger = logging.getLogger(__name__)
self.config = self.load_config(config_file)
self.sync_history = []
def load_config(self, config_file: str) -> Dict:
"""加载同步配置"""
with open(config_file, 'r', encoding='utf-8') as f:
return json.load(f)
def sync_module(self, module_name: str, sync_type: str = 'incremental'):
"""同步指定模块
Args:
module_name: 模块名称
sync_type: 同步类型 (full/incremental)
"""
if module_name not in self.config['modules']:
self.logger.error(f"未找到模块: {module_name}")
return False
module_config = self.config['modules'][module_name]
# 获取数据源
source_data = self._get_source_data(module_config, sync_type)
# 数据转换
transformed_data = self._transform_data(source_data, module_config)
# 写入目标
success = self._write_to_target(transformed_data, module_config)
# 记录同步历史
self._record_sync_history(module_name, sync_type, success)
return success
def _get_source_data(self, config: Dict, sync_type: str) -> List[Dict]:
"""获取源数据"""
source_type = config.get('source_type', 'api')
if source_type == 'api':
return self._fetch_from_api(config['source_config'])
elif source_type == 'database':
return self._fetch_from_db(config['source_config'])
elif source_type == 'file':
return self._fetch_from_file(config['source_config'])
else:
raise ValueError(f"不支持的数据源类型: {source_type}")
def _fetch_from_api(self, config: Dict) -> List[Dict]:
"""从API获取数据"""
import requests
params = {
'page_size': config.get('page_size', 100),
'page': 1
}
# 增量同步时添加时间过滤
if config.get('last_sync_time'):
params['updated_after'] = config['last_sync_time']
all_data = []
while True:
response = requests.get(config['url'], params=params)
data = response.json()
all_data.extend(data.get('items', []))
if not data.get('has_more'):
break
params['page'] += 1
return all_data
def _transform_data(self, data: List[Dict], config: Dict) -> List[Dict]:
"""数据转换"""
mapping = config.get('field_mapping', {})
transformed = []
for item in data:
new_item = {}
for source_field, target_field in mapping.items():
new_item[target_field] = item.get(source_field)
transformed.append(new_item)
return transformed
def _write_to_target(self, data: List[Dict], config: Dict) -> bool:
"""写入目标系统"""
target_type = config.get('target_type', 'database')
if target_type == 'database':
return self._write_to_db(data, config['target_config'])
elif target_type == 'file':
return self._write_to_file(data, config['target_config'])
elif target_type == 'redis':
return self._write_to_redis(data, config['target_config'])
else:
raise ValueError(f"不支持的目标类型: {target_type}")
def _record_sync_history(self, module: str, sync_type: str, success: bool):
"""记录同步历史"""
self.sync_history.append({
'module': module,
'sync_type': sync_type,
'timestamp': datetime.now().isoformat(),
'success': success
})
# 配置文件示例 (sync_config.json)
"""
{
"modules": {
"user": {
"source_type": "api",
"source_config": {
"url": "https://api.example.com/users",
"page_size": 100
},
"target_type": "database",
"target_config": {
"host": "localhost",
"database": "business_db",
"table": "users"
},
"field_mapping": {
"id": "user_id",
"name": "username",
"email": "email"
}
},
"order": {
"source_type": "api",
"source_config": {
"url": "https://api.example.com/orders",
"page_size": 50
},
"target_type": "file",
"target_config": {
"path": "/data/orders/",
"format": "json"
},
"field_mapping": {
"order_id": "id",
"total_amount": "amount"
}
}
}
}
"""
# 使用示例
sync = BusinessDataSync('sync_config.json')
# 按需同步用户模块
sync.sync_module('user', 'incremental')
# 全量同步订单模块
sync.sync_module('order', 'full')
带依赖管理的同步方案
from typing import Dict, List, Set
from enum import Enum
class SyncStatus(Enum):
PENDING = 'pending'
SYNCING = 'syncing'
COMPLETED = 'completed'
FAILED = 'failed'
class DependencySyncManager:
"""依赖同步管理器"""
def __init__(self):
self.dependencies = {
'user': [], # 用户模块无依赖
'order': ['user'], # 订单模块依赖用户
'product': ['category'], # 产品模块依赖分类
'category': [], # 分类模块无依赖
'inventory': ['product', 'warehouse'], # 库存依赖产品和仓库
'warehouse': []
}
self.module_status = {}
self.sync_handlers = {}
def register_handler(self, module: str, handler_func):
"""注册同步处理器"""
self.sync_handlers[module] = handler_func
def get_sync_order(self, modules: List[str]) -> List[str]:
"""获取同步顺序（拓扑排序）"""
from collections import deque
# 计算入度
in_degree = {m: 0 for m in modules}
graph = {m: [] for m in modules}
for module in modules:
for dep in self.dependencies.get(module, []):
if dep in modules:
in_degree[module] = in_degree.get(module, 0) + 1
graph[dep].append(module)
# BFS拓扑排序
queue = deque([m for m in modules if in_degree[m] == 0])
sorted_modules = []
while queue:
module = queue.popleft()
sorted_modules.append(module)
for next_module in graph[module]:
in_degree[next_module] -= 1
if in_degree[next_module] == 0:
queue.append(next_module)
# 检查是否有循环依赖
if len(sorted_modules) != len(modules):
raise ValueError("检测到循环依赖关系")
return sorted_modules
def sync_modules_with_dependencies(self, modules: List[str]):
"""同步指定模块及其依赖"""
# 获取所有需要同步的模块（包括依赖）
modules_to_sync = set()
for module in modules:
modules_to_sync.add(module)
modules_to_sync.update(self._get_all_dependencies(module))
# 获取同步顺序
sync_order = self.get_sync_order(list(modules_to_sync))
results = {}
print(f"同步顺序: {sync_order}")
for module in sync_order:
print(f"正在同步模块: {module}")
self.module_status[module] = SyncStatus.SYNCING
try:
if module in self.sync_handlers:
result = self.sync_handlers[module]()
results[module] = result
self.module_status[module] = SyncStatus.COMPLETED
print(f"模块 {module} 同步完成")
else:
raise ValueError(f"未找到模块 {module} 的同步处理器")
except Exception as e:
self.module_status[module] = SyncStatus.FAILED
print(f"模块 {module} 同步失败: {e}")
results[module] = None
return results
def _get_all_dependencies(self, module: str) -> Set[str]:
"""获取模块的所有依赖（递归）"""
dependencies = set()
for dep in self.dependencies.get(module, []):
if dep not in dependencies:
dependencies.add(dep)
dependencies.update(self._get_all_dependencies(dep))
return dependencies
# 使用示例
def sync_user_module():
print("执行用户数据同步...")
return {"synced_records": 100}
def sync_order_module():
print("执行订单数据同步...")
return {"synced_records": 200}
def sync_product_module():
print("执行产品数据同步...")
return {"synced_records": 50}
# 初始化管理器
manager = DependencySyncManager()
manager.register_handler('user', sync_user_module)
manager.register_handler('order', sync_order_module)
manager.register_handler('product', sync_product_module)
manager.register_handler('category', lambda: {"synced_records": 30})
# 按需同步（自动处理依赖）
results = manager.sync_modules_with_dependencies(['order', 'product'])
print(f"同步结果: {results}")
带缓存和断点续传的同步方案
import pickle
import hashlib
from datetime import datetime, timedelta
class CacheSyncManager:
"""带缓存的同步管理器"""
def __init__(self, cache_dir: str = './sync_cache'):
import os
self.cache_dir = cache_dir
os.makedirs(cache_dir, exist_ok=True)
self.cache_ttl = {
'user': timedelta(hours=1),
'order': timedelta(minutes=30),
'product': timedelta(days=1)
}
def get_cache_key(self, module: str, params: Dict = None) -> str:
"""生成缓存键"""
base_key = f"sync_{module}"
if params:
param_str = json.dumps(params, sort_keys=True)
base_key = f"{base_key}_{hashlib.md5(param_str.encode()).hexdigest()[:8]}"
return base_key
def get_cached_data(self, module: str, params: Dict = None) -> Optional[List[Dict]]:
"""获取缓存数据"""
cache_key = self.get_cache_key(module, params)
cache_file = f"{self.cache_dir}/{cache_key}.pkl"
try:
with open(cache_file, 'rb') as f:
cached_data = pickle.load(f)
# 检查缓存有效期
if datetime.now() - cached_data['timestamp'] < self.cache_ttl.get(module, timedelta(hours=1)):
return cached_data['data']
except (FileNotFoundError, pickle.UnpicklingError):
pass
return None
def set_cached_data(self, module: str, data: List[Dict], params: Dict = None):
"""设置缓存数据"""
cache_key = self.get_cache_key(module, params)
cache_file = f"{self.cache_dir}/{cache_key}.pkl"
cached_data = {
'timestamp': datetime.now(),
'data': data,
'module': module
}
with open(cache_file, 'wb') as f:
pickle.dump(cached_data, f)
def sync_with_checkpoint(self, module: str, batch_size: int = 100):
"""带检查点的同步（支持断点续传）"""
checkpoint_file = f"{self.cache_dir}/checkpoint_{module}.json"
# 检查是否有检查点
checkpoint = self._load_checkpoint(checkpoint_file)
start_id = checkpoint.get('last_processed_id', 0) if checkpoint else 0
print(f"从ID {start_id} 继续同步模块 {module}")
processed_count = 0
has_more = True
while has_more:
try:
# 分批获取数据
batch_data = self._fetch_batch(module, start_id, batch_size)
if not batch_data:
has_more = False
break
# 处理批次数据
self._process_batch(module, batch_data)
# 更新检查点
last_id = batch_data[-1].get('id')
self._save_checkpoint(checkpoint_file, {
'last_processed_id': last_id,
'processed_count': processed_count + len(batch_data),
'last_update': datetime.now().isoformat()
})
processed_count += len(batch_data)
start_id = last_id
print(f"已同步 {processed_count} 条记录")
except Exception as e:
print(f"同步中断于ID {start_id}: {e}")
return {
'success': False,
'processed_count': processed_count,
'next_checkpoint': start_id
}
# 同步完成，删除检查点
import os
os.remove(checkpoint_file)
return {
'success': True,
'processed_count': processed_count
}
def _load_checkpoint(self, filepath: str) -> Optional[Dict]:
"""加载检查点"""
try:
with open(filepath, 'r') as f:
return json.load(f)
except FileNotFoundError:
return None
def _save_checkpoint(self, filepath: str, checkpoint: Dict):
"""保存检查点"""
with open(filepath, 'w') as f:
json.dump(checkpoint, f)
def _fetch_batch(self, module: str, start_id: int, batch_size: int) -> List[Dict]:
"""分批获取数据"""
# 模拟数据库查询
import random
return [
{'id': start_id + i, 'data': f'data_{i}'}
for i in range(batch_size)
if random.random() > 0.1 # 模拟数据终止
]
def _process_batch(self, module: str, batch_data: List[Dict]):
"""处理批次数据"""
# 实际应用中的业务处理
pass
# 使用示例
cache_manager = CacheSyncManager()
# 检查缓存
cached_users = cache_manager.get_cached_data('user')
if cached_users:
print("使用缓存数据")
else:
print("缓存过期或不存在，执行同步")
# 执行同步逻辑
new_data = [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]
cache_manager.set_cached_data('user', new_data)
# 断点续传同步
result = cache_manager.sync_with_checkpoint('order', batch_size=50)
print(f"同步结果: {result}")
命令行调用示例
import argparse
import sys
def main():
parser = argparse.ArgumentParser(description='业务数据同步工具')
parser.add_argument('--modules', nargs='+', required=True, help='需要同步的模块')
parser.add_argument('--sync-type', choices=['full', 'incremental'], default='incremental')
parser.add_argument('--config', default='sync_config.json', help='配置文件路径')
parser.add_argument('--parallel', action='store_true', help='是否并行同步')
parser.add_argument('--force', action='store_true', help='强制重新同步')
args = parser.parse_args()
# 初始化同步器
from concurrent.futures import ThreadPoolExecutor
if args.parallel:
# 并行同步
with ThreadPoolExecutor(max_workers=5) as executor:
futures = []
for module in args.modules:
future = executor.submit(
sync_module_wrapper,
module,
args.sync_type,
args.config,
args.force
)
futures.append(future)
for future in futures:
result = future.result()
print(f"同步结果: {result}")
else:
# 串行同步
for module in args.modules:
result = sync_module_wrapper(module, args.sync_type, args.config, args.force)
print(f"模块 {module} 同步 {'成功' if result else '失败'}")
def sync_module_wrapper(module, sync_type, config, force):
"""同步模块包装函数"""
sync = BusinessDataSync(config)
return sync.sync_module(module, sync_type)
if __name__ == '__main__':
main()
使用命令行调用：
# 同步指定模块 python sync.py --modules user order product # 全量同步 python sync.py --modules user --sync-type full # 并行同步 python sync.py --modules user order product --parallel # 强制重新同步 python sync.py --modules user --force
使用建议
-
选择合适的同步策略：
- 全量同步：首次同步或数据一致性要求高时使用
- 增量同步：日常数据更新，效率更高
- 基于时间戳或版本号的增量同步
-
错误处理：
- 实现重试机制,支持断点续传
- 记录同步日志,便于排查问题
- 设置同步超时时间
-
性能优化：
- 使用缓存减少重复同步
- 批量处理大数量数据
- 考虑并行同步非依赖模块
-
监控告警：
- 监控同步状态和延迟
- 同步失败时发送通知
- 定期检查数据一致性
这些方案可以根据你的具体业务需求进行调整和扩展。
