本文目录导读:

我来详细介绍Python脚本适配定时周期同步场景的几种方案:
使用 schedule 库(简单易用)
import schedule
import time
from datetime import datetime
def sync_data():
"""数据同步函数"""
print(f"[{datetime.now()}] 开始数据同步...")
# 你的同步逻辑
print(f"[{datetime.now()}] 同步完成")
# 每隔5分钟执行一次
schedule.every(5).minutes.do(sync_data)
# 每天在特定时间执行
schedule.every().day.at("10:30").do(sync_data)
# 每周一执行
schedule.every().monday.do(sync_data)
# 每小时执行
schedule.every().hour.do(sync_data)
# 每10秒执行一次
schedule.every(10).seconds.do(sync_data)
if __name__ == "__main__":
while True:
schedule.run_pending()
time.sleep(1)
使用 APScheduler(功能强大)
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger
from datetime import datetime
import logging
# 配置日志
logging.basicConfig(level=logging.INFO)
def sync_data_job():
"""数据同步任务"""
logging.info(f"[{datetime.now()}] 执行数据同步")
# 同步逻辑
logging.info(f"[{datetime.now()}] 数据同步完成")
def create_scheduler():
"""创建调度器"""
scheduler = BlockingScheduler()
# 方式1:间隔触发 - 每5分钟执行一次
scheduler.add_job(
sync_data_job,
IntervalTrigger(minutes=5),
id='sync_5min',
name='每5分钟同步'
)
# 方式2:Cron触发 - 每天凌晨2点执行
scheduler.add_job(
sync_data_job,
CronTrigger(hour=2, minute=0),
id='sync_daily',
name='每日同步'
)
# 方式3:每小时的第30分钟执行
scheduler.add_job(
sync_data_job,
CronTrigger(minute=30),
id='sync_hourly',
name='每小时同步'
)
# 方式4:复杂Cron表达式(周一到周五每小时执行)
scheduler.add_job(
sync_data_job,
CronTrigger(day_of_week='mon-fri', hour='9-18', minute=0),
id='sync_work_hours',
name='工作时间同步'
)
return scheduler
if __name__ == "__main__":
scheduler = create_scheduler()
try:
scheduler.start()
except KeyboardInterrupt:
scheduler.shutdown()
使用 time.sleep(简单直接)
import time
from datetime import datetime
def sync_data():
"""数据同步函数"""
print(f"[{datetime.now()}] 开始数据同步...")
# 同步逻辑
print(f"[{datetime.now()}] 同步完成")
def run_periodic_sync(interval_seconds=300):
"""定时执行同步"""
next_run = time.time()
while True:
current_time = time.time()
if current_time >= next_run:
sync_data()
next_run = current_time + interval_seconds
time.sleep(1) # 每秒检查一次
if __name__ == "__main__":
# 每5分钟执行一次
run_periodic_sync(300)
高级版:带重试和错误处理的同步器
import time
from datetime import datetime, timedelta
import logging
from functools import wraps
import traceback
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class PeriodicSyncManager:
"""定时同步管理器"""
def __init__(self, interval_seconds=300):
self.interval = interval_seconds
self.running = False
self.last_sync_time = None
self.sync_count = 0
self.error_count = 0
def retry_on_failure(max_retries=3, delay_seconds=10):
"""重试装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
logger.error(f"执行失败 (尝试 {attempt + 1}/{max_retries}): {e}")
if attempt < max_retries - 1:
logger.info(f"{delay_seconds}秒后重试...")
time.sleep(delay_seconds)
else:
raise
return wrapper
return decorator
@retry_on_failure(max_retries=3)
def sync_data(self):
"""数据同步核心逻辑"""
logger.info(f"开始第 {self.sync_count + 1} 次同步...")
# 模拟同步逻辑
# 这里替换为你的实际同步代码
import random
if random.random() < 0.2: # 20%概率失败
raise Exception("模拟同步失败")
self.last_sync_time = datetime.now()
self.sync_count += 1
logger.info(f"同步完成 - 第 {self.sync_count} 次")
def sync_with_monitoring(self):
"""带监控的同步"""
try:
start_time = time.time()
self.sync_data()
elapsed = time.time() - start_time
# 记录同步指标
metrics = {
'sync_time': elapsed,
'last_sync': self.last_sync_time,
'total_syncs': self.sync_count,
'errors': self.error_count
}
# 检查同步时间是否过长(超过间隔的80%)
if elapsed > self.interval * 0.8:
logger.warning(f"同步耗时过长: {elapsed:.2f}秒")
return metrics
except Exception as e:
self.error_count += 1
logger.error(f"同步失败: {e}")
logger.debug(traceback.format_exc())
return None
def run(self):
"""主运行循环"""
self.running = True
logger.info(f"定时同步管理器启动 - 间隔: {self.interval}秒")
# 立即执行第一次同步
self.sync_with_monitoring()
next_sync_time = time.time() + self.interval
while self.running:
try:
current_time = time.time()
if current_time >= next_sync_time:
self.sync_with_monitoring()
next_sync_time = current_time + self.interval
# 动态调整检查频率
sleep_time = min(1, (next_sync_time - time.time()) / 2)
time.sleep(max(0.1, sleep_time))
except KeyboardInterrupt:
logger.info("收到停止信号")
break
except Exception as e:
logger.error(f"运行异常: {e}")
time.sleep(5)
self.shutdown()
def shutdown(self):
"""关闭管理器"""
self.running = False
logger.info("定时同步管理器已停止")
logger.info(f"总计同步: {self.sync_count}次, 失败: {self.error_count}次")
if __name__ == "__main__":
# 创建并运行同步管理器
manager = PeriodicSyncManager(interval_seconds=60) # 每分钟同步一次
manager.run()
配置文件驱动的同步器
# sync_config.yaml
sync_jobs:
- name: "business_data_sync"
enabled: true
schedule:
type: "interval"
minutes: 5
retry:
max_retries: 3
delay_seconds: 10
- name: "daily_backup"
enabled: true
schedule:
type: "cron"
hour: 2
minute: 0
retry:
max_retries: 2
delay_seconds: 30
import yaml
from typing import Dict, Any
class ConfigDrivenSync:
"""配置驱动的同步器"""
def __init__(self, config_path: str):
with open(config_path, 'r') as f:
self.config = yaml.safe_load(f)
self.jobs = {}
self.setup_jobs()
def setup_jobs(self):
"""根据配置设置定时任务"""
for job_config in self.config['sync_jobs']:
if not job_config.get('enabled', False):
continue
job_name = job_config['name']
self.jobs[job_name] = {
'config': job_config,
'last_run': None,
'next_run': datetime.now()
}
def run(self):
"""执行同步循环"""
while True:
current_time = datetime.now()
for job_name, job_info in self.jobs.items():
if current_time >= job_info['next_run']:
self.execute_job(job_name)
time.sleep(1)
def execute_job(self, job_name: str):
"""执行指定任务"""
# 执行逻辑
pass
最佳实践建议
-
选择合适的方案
- 简单场景:
time.sleep或schedule - 企业级应用:
APScheduler - 分布式系统:考虑 Celery 或 RQ
- 简单场景:
-
错误处理
- 实现重试机制
- 设置最大重试次数
- 记录失败日志
-
监控与告警
- 记录同步状态和指标
- 设置同步时间过长告警
- 统计成功率
-
性能优化
- 避免频繁的同步
- 实现增量同步
- 使用连接池
-
部署建议
- 使用 systemd 或 supervisor 管理进程
- 容器化部署
- 考虑高可用方案
选择哪种方案取决于你的具体需求和项目复杂度,对于大多数商业场景,推荐使用 APScheduler,它提供了最全面的功能。