Python 定时任务实战：用 APScheduler 构建可靠的每日作业调度系统

一、为什么需要“可靠”的定时任务

几乎每个后端系统里都存在“每天在固定时间做某件事”的需求：每天凌晨清理临时文件、每天 9 点生成并推送报表、每小时同步一次第三方数据……这类需求看起来简单，但真正上线后，最容易出问题的恰恰是它们——服务重启任务丢了、多进程下任务被执行了 N 次、某次执行超时后线程被卡死、时区没对齐导致任务提前/延后一小时。

本文从工程实践角度，讲清楚如何用 Python 的 APScheduler 构建一套“可持久化、可并发控制、可观测、可守护”的每日作业调度系统，并给出部署到华为云 ECS 的完整方案。

二、选型：为什么不直接用 cron / while + sleep

• 系统 cron：稳定、简单，但任务配置散落在机器上，难以随业务代码一起版本化；容器环境下不友好；也不方便做“任务失败重试、错过后补跑”这类逻辑。
• while True + sleep：最原始，进程挂掉任务就没了，且 sleep 会造成时间漂移（每轮都会累积误差）。
• APScheduler：纯 Python、进程内调度，支持 cron/interval/date 三种触发器、任务持久化、并发控制、任务监听与异常回调，能和业务代码放在同一个仓库里，部署灵活。

对于“单机、中小规模、需要和业务代码强绑定”的场景，APScheduler 是非常合适的选择。

三、快速上手

安装：

pip install apscheduler

最小示例：

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

def daily_report():
    print("生成日报:", datetime.now())

sched = BlockingScheduler(timezone="Asia/Shanghai")
sched.add_job(daily_report, "cron", hour=9, minute=0, id="daily_report")
sched.start()   # 阻塞当前进程

上面的任务会在每天 09:00（上海时区）执行一次。id 建议显式指定，便于后续修改、暂停、删除任务。

四、三种触发器怎么选

1. date：在某个具体时间点执行一次，适合“延时任务”。

from datetime import datetime, timedelta
sched.add_job(do_once, "date", run_date=datetime.now() + timedelta(minutes=30))

2. interval：固定间隔执行，适合轮询、心跳。

sched.add_job(sync_data, "interval", minutes=10)

3. cron：类 cron 表达式，适合固定时刻。

sched.add_job(daily_report, "cron", day_of_week="mon-fri", hour=9, minute=0)

注意 APScheduler 的 day_of_week 用 0-6 表示周一到周日，hour 是 0-23，和 Linux cron 的“分钟在前”不同，别混用。

五、任务持久化：别让重启丢任务

默认情况下 APScheduler 把任务存在内存里，进程一重启，动态添加的任务就全没了。生产环境务必配置 SQLAlchemyJobStore：

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore

jobstores = {
    "default": SQLAlchemyJobStore(url="sqlite:///jobs.sqlite")
}
sched = BlockingScheduler(jobstores=jobstores, timezone="Asia/Shanghai")

把 url 换成你的数据库连接串即可（生产建议用独立的库或表前缀，避免和业务表混在一起）。需要注意：被持久化的任务函数必须是可导入的顶层函数，不能是 lambda 或局部函数，否则反序列化会失败。

六、并发控制：避免任务“叠罗汉”

如果某个任务执行时间超过了调度间隔，默认会不断堆叠新实例，最终把线程池占满。通过 job_defaults 统一约束：

from apscheduler.executors.pool import ThreadPoolExecutor

executors = {"default": ThreadPoolExecutor(10)}
job_defaults = {
    "coalesce": True,          # 错过的多次执行合并为一次
    "max_instances": 1,        # 同一任务最多并发 1 个实例
    "misfire_grace_time": 60,  # 错过 60s 内仍允许补跑
}
sched = BlockingScheduler(executors=executors, job_defaults=job_defaults,
                          timezone="Asia/Shanghai")

这三项基本能解决 90% 的“任务重复执行/堆积”事故。

七、异常处理与可观测性

调度器要能“自己报告问题”。推荐在任务内部统一 try/except，并加一个监听器记录执行结果：

from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR

def on_job_event(event):
    if event.exception:
        print(f"[任务异常] {event.job_id}: {event.exception}")
    else:
        print(f"[任务完成] {event.job_id}")

sched.add_listener(on_job_event, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)

配合结构化日志（logging/json），再把日志采集到华为云 LTS（日志服务），就能对“今天任务有没有跑、跑了多久、有没有报错”一目了然。

八、部署到华为云 ECS：用 systemd 守护

调度进程需要一个“看门狗”保证挂了自动拉起。在华为云 ECS 上用 systemd 是最省心的方式：

[Unit]
Description=Daily Scheduler
After=network.target

[Service]
Type=simple
WorkingDirectory=/opt/app
ExecStart=/usr/bin/python3 /opt/app/scheduler.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

保存为 /etc/systemd/system/daily-scheduler.service，然后：

sudo systemctl daemon-reload
sudo systemctl enable --now daily-scheduler
sudo systemctl status daily-scheduler

这样即使进程崩溃、ECS 重启，调度器也会自动恢复，任务持久化 + 自动拉起，基本做到“不用盯着”。

九、几个常见的坑

1. 时区：一定显式设置 timezone="Asia/Shanghai"，否则容器里默认 UTC，任务会差 8 小时。
2. 多进程重复执行：如果服务用 gunicorn/uwsgi 起了多个 worker，每个 worker 都跑一个调度器，任务会被执行多次。调度器应该是单进程的，和 Web 服务分离部署。
3. 时间漂移：用 interval 时睡眠会累积误差，对时间敏感的任务优先用 cron。
4. 持久化函数可导入：前面提过，lambda/闭包无法反序列化，务必用顶层函数。

十、总结

一个“能打”的定时任务系统 = 合适的触发器 + 任务持久化 + 并发约束 + 异常可观测 + 进程守护。APScheduler 把前四件事都封装好了，剩下的部署交给华为云 ECS + systemd 即可。把这套模板沉淀下来，后续所有“每日/每小时”的需求都能低成本复用，把重复劳动交给机器，把精力留给真正有价值的业务逻辑。