本文目录导读：
我来介绍几种简化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()
这些方法可以帮助你简化低频同步执行流程：
- 装饰器：简化重试和定时逻辑
- 调度器：管理定时任务
- 上下文管理器：管理资源和状态
- 状态机：控制执行流程
- 配置文件：分离配置和代码
选择哪种方式取决于你的具体需求：
- 简单重试 → 装饰器
- 定时任务 → 调度器
- 资源管理 → 上下文管理器
- 流程控制 → 状态机
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
