CodeWalk

APScheduler：高级定时任务框架详解

作者：我是大山 · 2026-05-30 12:55

请介绍APScheduler（Advanced Python Scheduler）的架构（触发器/作业存储/执行器/调度器）、三种触发器类型（date/interval/cron）、以及作业存储（内存/SQLite/Redis/MongoDB）。给出一个用APScheduler实现定时任务的完整示例。

回答

我是大山

APScheduler架构

四个核心组件：

  1. 触发器（Trigger）：决定作业何时执行
  2. 作业存储（Job Store）：持久化作业状态
  3. 执行器（Executor）：在线程/进程池中执行作业
  4. 调度器（Scheduler）：协调上述组件

三种触发器

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.date import DateTrigger
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.cron import CronTrigger

scheduler = BackgroundScheduler()

# 1. DateTrigger — 指定时间执行一次
scheduler.add_job(
    func=my_job,
    trigger=DateTrigger(run_date='2026-06-01 10:00:00'),
    id='once'
)

# 2. IntervalTrigger — 间隔执行
scheduler.add_job(
    func=my_job,
    trigger=IntervalTrigger(minutes=30),
    id='interval',
    replace_existing=True
)

# 3. CronTrigger — 类似crontab
scheduler.add_job(
    func=my_job,
    trigger=CronTrigger(
        day_of_week='mon-fri',
        hour=9,
        minute=30
    ),
    id='cron'
)

scheduler.start()

作业存储

存储类型持久化适用场景
MemoryJobStore不持久化开发测试
SQLAlchemyJobStoreSQLite/MySQL/PostgreSQL中等规模
MongoDBJobStoreMongoDB高可用分布式
RedisJobStoreRedis高性能内存

完整示例

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
import logging

logging.basicConfig(level=logging.INFO)

def daily_report():
    print('生成日报...')

scheduler = BackgroundScheduler()
scheduler.add_job(
    daily_report,
    CronTrigger(hour=8, minute=0),
    id='daily_report',
    misfire_grace_time=300  # 错过执行后5分钟内补执行
)

scheduler.start()

try:
    while True:
        pass
except KeyboardInterrupt:
    scheduler.shutdown()

注意： APScheduler不是生产级分布式调度器（无Leader选举），分布式场景考虑Celery Beat。