【Python】APScheduler：Python中强大的任务调度库

我听见有人猜
你是敌人潜伏的内线
和你相知多年
我确信对你的了解
你舍命救我画面
一一在眼前浮现
司空见惯了鲜血
你忘记你本是娇娆的红颜
感觉你我彼此都那么依恋
🎵 许嵩《内线》

在 Python 开发中，定时任务是非常常见的需求，例如定期备份数据库、定时发送邮件提醒，或按固定时间抓取数据。如果你需要一个功能强大、灵活的调度库，APScheduler（Advanced Python Scheduler）是一个非常好的选择。

APScheduler 不仅支持简单的定时任务，还支持复杂的调度需求。它能够基于日期、固定时间间隔、或 cron 表达式调度任务，并且可以与多种后端存储系统（如数据库、Redis）集成，实现任务的持久化管理。

1. 什么是 APScheduler？

APScheduler 是一个功能强大的 Python 任务调度库。它支持以下几种任务调度方式：

基于时间间隔：每隔固定的时间执行任务。
基于日期：在特定的日期和时间点执行任务。
基于 cron 表达式：类似于 Linux 系统中的 cron 作业，可以非常灵活地指定复杂的时间调度。
持久化：支持任务存储到数据库中，重启后可以恢复任务。
APScheduler 还支持多种调度器，包括阻塞调度器（适合单线程的应用）、后台调度器（适合 Web 框架或多线程环境）、以及基于 Tornado、Twisted 等异步框架的调度器。

2. 安装 APScheduler

你可以通过 pip 来安装 APScheduler：

pip install apscheduler

安装完成后，你就可以开始在 Python 项目中使用 APScheduler 来调度任务。

3. 基本使用

APScheduler 使用起来非常灵活，常见的调度器是 BlockingScheduler 和 BackgroundScheduler。BlockingScheduler 通常用于简单脚本，它会阻塞主线程直到任务调度结束。而 BackgroundScheduler 可以在后台运行，适合与其他程序逻辑同时工作。

3.1 简单示例

下面是一个简单的任务调度示例，使用 BlockingScheduler 每隔 5 秒执行一次任务：

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetimedef job():print(f"任务执行时间：{datetime.now()}")scheduler = BlockingScheduler()# 每 5 秒执行一次
scheduler.add_job(job, 'interval', seconds=5)# 启动调度器
scheduler.start()

在这个例子中，调度器会每隔 5 秒执行一次 job() 函数。scheduler.add_job() 方法用于添加一个任务，第一个参数是任务函数，第二个参数是调度方式（此处为 ‘interval’，表示基于时间间隔的调度），后面的 seconds=5 表示时间间隔为 5 秒。

3.2 使用 BackgroundScheduler

如果你不想让调度器阻塞主线程，使用 BackgroundScheduler 可能更合适。BackgroundScheduler 可以在后台运行，适用于与其他程序逻辑并行运行的场景。

from apscheduler.schedulers.background import BackgroundScheduler
import timedef job():print(f"任务执行中...")scheduler = BackgroundScheduler()
scheduler.add_job(job, 'interval', seconds=5)# 启动调度器
scheduler.start()# 主程序继续执行
while True:print("主程序运行中...")time.sleep(3)

在这个例子中，BackgroundScheduler 启动后，主程序会继续运行，同时后台任务每 5 秒执行一次。

4. 基于不同触发器的调度

APScheduler 具有多种任务触发器，可以灵活地根据时间进行任务调度。常见的触发器有三种：

DateTrigger：在特定的日期时间执行任务。
IntervalTrigger：每隔固定的时间间隔执行任务。
CronTrigger：使用类似 cron 表达式的方式进行调度。

4.1 基于日期的调度（DateTrigger）

如果你需要在特定的时间点执行任务，可以使用 DateTrigger：

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime, timedeltadef job():print("一次性任务执行")scheduler = BlockingScheduler()# 设置任务在 5 秒后执行一次
run_time = datetime.now() + timedelta(seconds=5)
scheduler.add_job(job, 'date', run_date=run_time)scheduler.start()

在这个示例中，任务将在当前时间 5 秒后执行一次。

4.2 基于时间间隔的调度（IntervalTrigger）

间隔调度是 APScheduler 中最常用的调度方式之一。它允许你按照指定的时间间隔执行任务，例如每隔几秒、几分钟、几小时等。

scheduler.add_job(job, 'interval', minutes=10)  # 每 10 分钟执行一次

4.3 使用 Cron 表达式进行调度（CronTrigger）

CronTrigger 提供了最灵活的调度方式，类似于 Linux 系统中的 cron 作业。它允许你设置复杂的时间规则，如每周一上午 10:00 执行任务。

from apscheduler.schedulers.blocking import BlockingSchedulerdef job():print("每周任务执行")scheduler = BlockingScheduler()# 每周一上午 10 点执行
scheduler.add_job(job, 'cron', day_of_week='mon', hour=10, minute=0)scheduler.start()

这个例子展示了如何使用 cron 表达式来调度任务，它将在每周一上午 10:00 执行。

5. 任务的持久化

APScheduler 允许将任务存储在数据库等持久化存储中，使得任务可以在程序重启后继续执行。你可以选择 SQLite、MySQL、PostgreSQL 等数据库作为存储后端。

持久化调度任务的步骤大致如下：

安装数据库驱动，如 sqlite 或 mysqlclient。
配置 APScheduler 使用 SQLAlchemyJobStore 来管理任务。
下面是一个使用 SQLite 的例子：

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore# 创建一个持久化存储（SQLite）
jobstores = {'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}scheduler = BackgroundScheduler(jobstores=jobstores)def job():print("任务执行中...")# 每 10 秒执行一次任务
scheduler.add_job(job, 'interval', seconds=10)scheduler.start()

在这个例子中，任务存储在 jobs.sqlite 数据库中，重启程序后，APScheduler 会从数据库中恢复任务。

6. 任务的暂停、恢复与删除

APScheduler 还支持对任务进行动态管理，如暂停、恢复和删除任务。你可以通过任务的 ID 来操作任务：

# 添加任务并指定任务 ID
scheduler.add_job(job, 'interval', seconds=10, id='my_job')# 暂停任务
scheduler.pause_job('my_job')# 恢复任务
scheduler.resume_job('my_job')# 删除任务
scheduler.remove_job('my_job')

7. 完整示例

下面是一个 APScheduler 的完整示例，展示了如何使用 BackgroundScheduler 进行定时任务调度，并将任务持久化到 SQLite 数据库中：

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from datetime import datetime# 使用 SQLite 作为持久化存储
jobstores = {'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}scheduler = BackgroundScheduler(jobstores=jobstores)def send_email():print(f"发送提醒邮件，当前时间：{datetime.now()}")def backup_database():print(f"备份数据库，当前时间：{datetime.now()}")# 每天上午 8 点发送提醒邮件
scheduler.add_job(send_email, 'cron', hour=8, minute=0, id='email_job')# 每周日晚上 10 点备份数据库
scheduler.add_job(backup_database, 'cron', day_of_week='sun', hour=22, id='backup_job')scheduler.start()while True:pass  # 模拟程序一直运行