Python schedule库实现定时任务的避坑指南

import schedule
import time

def sync_data():
    print("开始同步数据...")
    # 模拟一些工作
    time.sleep(10)
    print("数据同步完成！")

# 每5分钟执行一次
schedule.every(5).minutes.do(sync_data)

while True:
    schedule.run_pending()
    time.sleep(1)

import threading
import schedule
import time
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

def run_threaded(job_func):
    """将任务函数放入线程中执行"""
    job_thread = threading.Thread(target=job_func)
    job_thread.start()
    # 注意：这里我们不调用 job_thread.join()

schedule.every(5).minutes.do(run_threaded, sync_data)

def sync_data():
    """数据同步任务"""
    try:
        logger.info("数据同步任务开始。")
        # 这里是你的核心业务逻辑，比如调用API、读写数据库等
        time.sleep(10)  # 模拟耗时操作
        # 模拟一个可能的错误
        # 1 / 0
        logger.info("数据同步任务成功完成。")
    except Exception as e:
        # 捕获所有异常，记录错误日志，避免线程异常导致整个程序退出
        logger.error(f"数据同步任务发生严重错误: {e}", exc_info=True)

def run_scheduler():
    """运行调度器的循环"""
    logger.info("定时任务调度器开始运行...")
    while True:
        try:
            schedule.run_pending()
            time.sleep(1)  # 避免CPU空转
        except KeyboardInterrupt:
            logger.info("接收到中断信号，调度器正在停止...")
            break
        except Exception as e:
            # 捕获调度器本身的未知异常，记录并继续运行
            logger.critical(f"调度器主循环发生未知异常: {e}", exc_info=True)
            time.sleep(60)  # 发生严重错误时，等待一分钟再重试循环

# 创建一个全局锁，用于保护共享资源
data_file_lock = threading.Lock()

def sync_data():
    try:
        logger.info("数据同步任务开始。")
        # 在操作共享文件前获取锁
        with data_file_lock:
            logger.info("获取到文件锁，开始写入数据...")
            time.sleep(5)  # 模拟文件操作
            logger.info("数据写入完成，释放文件锁。")
        logger.info("数据同步任务成功完成。")
    except Exception as e:
        logger.error(f"数据同步任务发生错误: {e}", exc_info=True)

"""
一个健壮的、基于 schedule 库的 Python 定时任务执行器。
适用于需要 7x24 小时运行的后台服务。
"""
import schedule
import time
import threading
import logging
from datetime import datetime

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger(__name__)

# 示例：用于保护共享资源的锁
shared_resource_lock = threading.Lock()

def run_threaded(job_func):
    """
    将任务函数放入独立线程中执行，避免阻塞主调度循环。
    Args:
        job_func: 要执行的任务函数
    """
    job_thread = threading.Thread(target=job_func, name=job_func.__name__)
    job_thread.start()
    # 重要：不在此处join，防止阻塞调度器。

def sync_data():
    """示例任务1：模拟数据同步"""
    task_name = "数据同步"
    try:
        logger.info(f"[{task_name}] 任务开始。")
        # 模拟一些网络I/O或数据库操作
        time.sleep(12)  # 假设本次执行需要12秒
        # 模拟操作共享资源
        with shared_resource_lock:
            logger.info(f"[{task_name}] 获取锁，操作共享资源...")
            time.sleep(1)
        # 模拟一个偶然的错误（取消注释来测试）
        # if datetime.now().second % 10 == 0:
        #     raise ConnectionError("模拟网络突然断开")
        logger.info(f"[{task_name}] 任务成功完成。")
    except Exception as e:
        logger.error(f"[{task_name}] 执行失败: {e}", exc_info=True)

def cleanup_temp():
    """示例任务2：模拟清理临时文件"""
    task_name = "清理临时文件"
    try:
        logger.info(f"[{task_name}] 任务开始。")
        time.sleep(3)  # 模拟清理工作
        logger.info(f"[{task_name}] 任务完成。")
    except Exception as e:
        logger.error(f"[{task_name}] 执行失败: {e}", exc_info=True)

def send_heartbeat():
    """示例任务3：发送心跳信号，证明调度器还活着"""
    try:
        # 这是一个非常快的任务，不需要放线程也行，这里为了统一也放了。
        logger.info("心跳检测：调度器运行正常。")
    except Exception as e:
        logger.error(f"心跳任务失败: {e}", exc_info=True)

def run_scheduler():
    """调度器主循环"""
    logger.info("=== 定时任务调度器启动 ===")

    # 1. 定义任务调度规则
    # 每5分钟执行一次数据同步（使用线程）
    schedule.every(5).minutes.do(run_threaded, sync_data)
    # 每小时执行一次清理
    schedule.every().hour.do(run_threaded, cleanup_temp)
    # 每30秒发送一次心跳
    schedule.every(30).seconds.do(run_threaded, send_heartbeat)

    # 2. 启动后立即运行一次某些任务（可选）
    logger.info("启动后立即执行一次清理和心跳...")
    run_threaded(cleanup_temp)
    run_threaded(send_heartbeat)

    # 3. 坚不可摧的主循环
    while True:
        try:
            schedule.run_pending()
            # 睡眠1秒，平衡CPU使用率和调度精度
            time.sleep(1)
        except KeyboardInterrupt:
            logger.info("接收到键盘中断信号，正在优雅停止调度器...")
            break
        except Exception as e:
            # 兜底：捕获调度循环自身的任何异常，记录并尝试恢复
            logger.critical(f"调度器主循环发生未预期异常，将在一分钟后重试: {e}", exc_info=True)
            time.sleep(60)

    logger.info("=== 定时任务调度器已停止 ===")

if __name__ == "__main__":
    run_scheduler()