本文目录导读：
我来介绍几种简化Python低频同步执行流程的方法：
使用装饰器简化重试逻辑
import time
from functools import wraps
def retry_on_failure(max_retries=3, delay=1):
"""自动重试装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_retries - 1:
raise
print(f"重试 {attempt + 1}/{max_retries}: {e}")
time.sleep(delay)
return None
return wrapper
return decorator
@retry_on_failure(max_retries=3)
def fetch_data():
"""模拟数据获取"""
import random
if random.random() < 0.7: # 70%概率失败
raise ConnectionError("网络错误")
return {"data": "成功获取"}
# 使用
result = fetch_data()
使用调度器简化定时执行
import schedule
import time
from datetime import datetime
class SimpleScheduler:
"""简化版调度器"""
def __init__(self):
self.jobs = []
def every(self, interval=1, unit='minutes'):
"""设置执行间隔"""
def decorator(func):
@wraps(func)
def wrapper():
try:
print(f"[{datetime.now()}] 执行任务: {func.__name__}")
return func()
except Exception as e:
print(f"任务失败: {e}")
# 根据单位设置调度
if unit == 'minutes':
schedule.every(interval).minutes.do(wrapper)
elif unit == 'hours':
schedule.every(interval).hours.do(wrapper)
elif unit == 'days':
schedule.every(interval).days.do(wrapper)
self.jobs.append(wrapper)
return wrapper
return decorator
def run(self):
"""启动调度器"""
print("调度器启动...")
while True:
schedule.run_pending()
time.sleep(1)
# 使用示例
scheduler = SimpleScheduler()
@scheduler.every(5, 'minutes')
def sync_data():
"""同步数据"""
print("正在同步数据...")
# 实际同步逻辑
return True
@scheduler.every(1, 'hours')
def clean_temp_files():
"""清理临时文件"""
print("清理临时文件...")
# 实际清理逻辑
# 启动调度器（如果需要）
# scheduler.run()
使用上下文管理器简化资源管理
from contextlib import contextmanager
import sqlite3
class SyncManager:
"""同步管理器"""
def __init__(self, db_path='sync.db'):
self.db_path = db_path
@contextmanager
def database_session(self):
"""数据库会话上下文管理器"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
try:
yield cursor
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
@contextmanager
def lock_file(self, lock_path='sync.lock'):
"""文件锁上下文管理器"""
import fcntl
lock_file = open(lock_path, 'w')
try:
fcntl.flock(lock_file, fcntl.LOCK_EX)
yield lock_file
finally:
fcntl.flock(lock_file, fcntl.LOCK_UN)
lock_file.close()
# 使用示例
manager = SyncManager()
# 使用数据库会话
with manager.database_session() as cursor:
cursor.execute("SELECT * FROM sync_records")
records = cursor.fetchall()
# 使用文件锁
with manager.lock_file():
print("执行互斥操作...")
使用简单的状态机模式
from enum import Enum
import time
class SyncState(Enum):
IDLE = "空闲"
RUNNING = "运行中"
SUCCESS = "成功"
FAILED = "失败"
class SimpleSyncMachine:
"""简化版同步状态机"""
def __init__(self):
self.state = SyncState.IDLE
self.last_sync_time = None
self.sync_count = 0
def sync(self):
"""执行同步操作"""
if self.state == SyncState.RUNNING:
print("同步正在执行中，跳过")
return False
self.state = SyncState.RUNNING
try:
# 模拟同步过程
self._do_sync()
self.state = SyncState.SUCCESS
self.last_sync_time = time.time()
self.sync_count += 1
return True
except Exception as e:
self.state = SyncState.FAILED
print(f"同步失败: {e}")
return False
def _do_sync(self):
"""实际的同步逻辑"""
print(f"执行第 {self.sync_count + 1} 次同步...")
time.sleep(1)
# 在这里添加实际的同步代码
def reset(self):
"""重置状态"""
self.state = SyncState.IDLE
print("状态已重置")
# 使用示例
sync = SimpleSyncMachine()
# 执行同步
sync.sync() # 成功执行
sync.sync() # 跳过，因为状态是 SUCCESS
sync.reset() # 重置状态
sync.sync() # 可以再次执行
完整示例：配置文件同步器
import json
import os
import hashlib
import shutil
from pathlib import Path
from datetime import datetime
class ConfigSync:
"""简化版配置文件同步器"""
def __init__(self, config_file='sync_config.json'):
self.config_file = config_file
self.config = self.load_config()
self.sync_records = {}
def load_config(self):
"""加载配置"""
if os.path.exists(self.config_file):
with open(self.config_file, 'r') as f:
return json.load(f)
return {
'source_dir': './source',
'backup_dir': './backup',
'sync_interval': 60, # 秒
'file_patterns': ['*.txt', '*.json', '*.yaml']
}
def calculate_md5(self, filepath):
"""计算文件MD5"""
hash_md5 = hashlib.md5()
with open(filepath, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""):
hash_md5.update(chunk)
return hash_md5.hexdigest()
def sync_files(self):
"""同步文件"""
source_path = Path(self.config['source_dir'])
backup_path = Path(self.config['backup_dir'])
if not source_path.exists():
print(f"源目录不存在: {source_path}")
return
backup_path.mkdir(parents=True, exist_ok=True)
for pattern in self.config['file_patterns']:
for filepath in source_path.glob(pattern):
if filepath.is_file():
self.sync_single_file(filepath, backup_path)
print(f"同步完成: {datetime.now()}")
def sync_single_file(self, source_file, backup_dir):
"""同步单个文件"""
backup_file = backup_dir / source_file.name
# 计算源文件MD5
source_md5 = self.calculate_md5(source_file)
# 检查备份文件
if backup_file.exists():
backup_md5 = self.calculate_md5(backup_file)
if source_md5 == backup_md5:
print(f"文件未变化: {source_file.name}")
return
# 备份文件
shutil.copy2(source_file, backup_file)
print(f"已同步: {source_file.name}")
# 记录同步信息
self.sync_records[source_file.name] = {
'time': str(datetime.now()),
'md5': source_md5
}
# 使用示例
sync = ConfigSync()
# 执行同步
sync.sync_files()
这些方法可以帮助你简化低频同步执行流程：
- 装饰器：简化重试和定时逻辑
- 调度器：管理定时任务
- 上下文管理器：管理资源和状态
- 状态机：控制执行流程
- 配置文件：分离配置和代码
选择哪种方式取决于你的具体需求：
- 简单重试 → 装饰器
- 定时任务 → 调度器
- 资源管理 → 上下文管理器
- 流程控制 → 状态机
