线程版 Cron 调度系统

# scheduler_system.py

from apscheduler.schedulers.background import BackgroundScheduler

from apscheduler.triggers.cron import CronTrigger

import threading

import time

# 示例任务函数

def task1():

    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 执行任务1 (每分钟执行一次)")

def task2():

    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 执行任务2 (每两分钟执行一次)")

# 初始化调度器（后台线程）

scheduler = BackgroundScheduler()

# 添加任务，使用 Cron 表达式

# Cron 表达式格式：second, minute, hour, day, month, day_of_week

scheduler.add_job(task1, CronTrigger.from_crontab('* * * * *'))      # 每分钟执行一次

scheduler.add_job(task2, CronTrigger.from_crontab('*/2 * * * *'))    # 每两分钟执行一次

# 启动调度器

scheduler.start()

print("调度系统已启动...")

# 主线程保持运行

try:

    while True:

        time.sleep(1)

except (KeyboardInterrupt, SystemExit):

    scheduler.shutdown()

    print("调度系统已停止。")

特点

Cron 表达式支持

APScheduler 支持标准 Cron 表达式：

*    *    *    *    *  

┬    ┬    ┬    ┬    ┬

│    │    │    │    │

│    │    │    │    └ day of week (0-6, 0=Sunday)

│    │    │    └───── month (1-12)

│    │    └────────── day of month (1-31)

│    └─────────────── hour (0-23)

└──────────────────── minute (0-59)

多任务并发执行

默认使用线程池执行任务，支持同时运行多个任务。

支持动态添加/移除任务

scheduler.add_job(my_task, CronTrigger.from_crontab('0 12 * * *'))  # 每天中午12点

scheduler.remove_job(job_id='my_job_id')  # 删除任务

后台运行，不阻塞主线程

可以把调度系统嵌入任何应用中，例如 Flask/Django、爬虫、自动化脚本等。

安装依赖

pip install apscheduler