APScheduler 使用指南

APScheduler 算是 Python 中一个比较轻量级的定时任务库了，基于 Quartz ，而且在项目上也用得不少（毕竟可以跨平台而且配置起来方便），之前接手项目的时候也了解过一下，太久没用又忘了 (ˉ▽ˉ；)… 想想如果在 Blog 里面记下来，有时候顺手翻翻也好。

这里介绍的 APScheduler 是 8 月份的 3.5.3 版本，本人用的是 Python 3.5。如果有别的问题，欢迎交流，或者查看 Version History ！

什么是 APScheduler

APScheduler ，全称是 Advanced Python Scheduler ，具体的介绍可以看 PyPI 或者 readthedocs 的文档介绍，这篇 blog 主要是翻译 User Guide 一节的主要内容，不过惯例还是先简单介绍一下这个库特别的地方。

APScheduler 内置了三种调度系统：

1. Linux Cron 风格的调度系统（并有可选的开始和结束时间）
2. 基于时间间隔的执行调度（周期性地运行作业 job ，并有可选的开始和结束时间）
3. 只执行一次的延后执行作业调度（只执行一次作业 job ，在设定的日期 date 或时间 time 执行）

APScheduler 可以配合多种不同的作业存储后端一起使用，目前支持以下的作业存储后端：

• 内存 Memory
• SQLAlchemy (任何 SQLAlchemy 支持的关系型数据库)
• MongoDB
• Redis
• RethinkDB
• ZooKeeper

APScheduler 也可以集成到几个常见的 Python 框架中，如：

• asyncio
• gevent
• Tornado
• Twisted
• Qt（使用 PyQt 或 PySide）

APScheduler 使用指南

代码示例

APScheduler 的源文件分发包里包含了 example 文件夹，在那里可以找到各种使用 APScheduler 的示例，这些示例同样可以查看在线版本。

基本概念

APScheduler 有如下四种组件：

• triggers 触发器:
  包含具体的角度逻辑。每个 job 都会有自己的触发器，由它来决定下一个要运行的 job 。在触发器被初始化配置之前，它们都是完全无状态（stateless）的。
• job stores 作业存储:
  存放被调度的 job 。默认的作业存储只是简单地将作业存储在内存中，但也可以存储到各种数据库中。当一个 job 保存到一个持久化地作业存储中时，其数据必须要被序列化（serialized），当它们被加载回来时再执行反序列化（deserialized）。非默认的作业存储不会将作业数据保存到内存中，相反，内存会作为后端存储介质在保存、加载、更新和搜索 job 过程中的中间人。作业存储不会在调度器（scheduler）之间共享。
• executors 执行器:
  负责处理运行中的作业。通常它们都是负责将 job 中指定的可调用的部分提交到线程或进程池。当 job 完成后，执行器会通知（notifies）调度器，由调度器随后发出（emits）一个恰当的事件（event）。
• schedulers 调度器:
  调度器负责将以上的东西结合在一起。一般情况下，你的应用程序只会有一个调度器在运行。应用程序的开发者通常不用直接面对 trigger ， job stores 以及 executor ，相反，调度器会提供合适的接口让开发者去管理它们 —— 通过调度程序来配置 job stores 和 executor 来实现诸如添加、修该和删除 job 。

如何选择合适的 scheduler、job stores、executor 和 trigger

scheduler 的选择取决于你程序的运行环境以及你想用 APScheduler 完成什么任务。这里有一份快速决定 scheduler 的指南：

• BlockingScheduler: 如果调度器是你程序中唯一要运行的东西，请选择它
• BackgroundScheduler: 如果你想你的调度器可以在你的应用程序后台静默运行，同时也不打算使用以下任何 Python 框架，请选择它
• AsyncIOScheduler: 如果你的程序使用了 asyncio 库，请使用这个调度器
• GeventScheduler: 如果你的程序使用了 gevent 库，请使用这个调度器
• TornadoScheduler: 如果你打算构建一个 Tornado 程序，请使用这个调度器
• TwistedScheduler: 如果你打算构建一个 Twisted 程序，请使用这个调度器
• QtScheduler: 如果你打算构建一个 Qt 程序，请使用这个调度器

为了选到合适的 job store ，你需要明确你是否需要将你的 job 持久化。如果你总是再应用程序开始的时候重新创建你的作业，那么你适合用默认的选项（MemoryJobStore）。但如果你需要持久化你的作业以面对 scheduler 重启或者应用程序崩溃的情况，那么你的选择通常需要考虑你在程序运行环境中所使用的工具。当然，如果你可以自由选择的话，我们建议使用 SQLAlchemyJobStore 配合 PostgreSQL 作为后端存储，因为这个组合提供了强大的数据完整性的保障。

同样的，executor 的选择基于你是否选择了以上任意一个 Python 框架。如果都没有，那么默认的 ThreadPoolExecutor 足够满足大部分的需求。如果你的作业包含了 CPU 密集型操作，你应该考虑使用 ProcessPoolExecutor 以便充分利用多核 CPU 。甚至你可以同时使用它们两者，将 process pool executor 作为备用 executor 。

当你调度一个 job 时，你需要为它设置一个 trigger 。trigger 将决定 job 何时运行。 APScheduler 有三个内置的 trigger 类型：

• date 在某个确定的时间点运行你的 job （只运行一次）
• interval 在固定的时间间隔周期性地运行你的 job
• cron 在一天的某些固定时间点周期性地运行你的 job

也可以将多个 trigger 组合在一起， job 的运行会在所有参与的 trigger 约定的时间点或者时任何一个满足条件的 trigger 时间点被触发。详情请看 combining triggers 文档。

你可以在 job store 、 executor 以及 trigger 各自的 API 文档页上找到对应的插件名称。

配置 scheduler

APScheduler 提供了许多不同的方法来配置 scheduler 。你可以使用一个配置字典，或者是直接将其作为 options 的关键字参数。你也可以 先实例化 scheduler ，随后再添加 job 和配置 scheduler 。后者可以为大多数环境提供最大的灵活性。

完整的调度器 层次配置选项（level configuration option）列表可以在 BaseScheduler 类的 API 引用页找到。

下面这个例子，使用默认的 job store 以及默认的 executor ，在你的应用程序中运行一个 BackgroundScheduler：

from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()

# 在这里可以初始化应用程序的剩余部分，当然也可以在初始化 scheduler 之前完成

下面是一个更加复杂而具体的例子：你有两个 job store 以及两个 executor ，同时要求调整新作业的默认值以设置不同的时区。以下的三段代码片段都是等价的。你会得到：

• 一个叫 mongo 的 MongoDBJobStore
• 一个叫 default 的 SQLAlchemyJobStore （使用 SQLite）
• 一个叫 default 的 ThreadPoolExecutor，使用 20 个工作线程
• 一个叫做 processpool 的 ProcessPoolExecutor，使用 5 个工作进程
• UTC 是调度器的时区
• 新 job 默认关闭聚合（coalescing）功能
• 每个新 job 默认限制最大实例数为 3

# Method 1
from pytz import utc

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.mongodb import MongoDBJobStore
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor


jobstores = {
    'mongo': MongoDBJobStore(),
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
executors = {
    'default': ThreadPoolExecutor(20),
    'processpool': ProcessPoolExecutor(5)
}
job_defaults = {
    'coalesce': False,
    'max_instances': 3
}
scheduler = BackgroundScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)

# Method 2
from apscheduler.schedulers.background import BackgroundScheduler


# 前缀 "apscheduler." 是硬编码的
scheduler = BackgroundScheduler({
    'apscheduler.jobstores.mongo': {
         'type': 'mongodb'
    },
    'apscheduler.jobstores.default': {
        'type': 'sqlalchemy',
        'url': 'sqlite:///jobs.sqlite'
    },
    'apscheduler.executors.default': {
        'class': 'apscheduler.executors.pool:ThreadPoolExecutor',
        'max_workers': '20'
    },
    'apscheduler.executors.processpool': {
        'type': 'processpool',
        'max_workers': '5'
    },
    'apscheduler.job_defaults.coalesce': 'false',
    'apscheduler.job_defaults.max_instances': '3',
    'apscheduler.timezone': 'UTC',
})

# Method 3
from pytz import utc

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ProcessPoolExecutor


jobstores = {
    'mongo': {'type': 'mongodb'},
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
executors = {
    'default': {'type': 'threadpool', 'max_workers': 20},
    'processpool': ProcessPoolExecutor(max_workers=5)
}
job_defaults = {
    'coalesce': False,
    'max_instances': 3
}
scheduler = BackgroundScheduler()

# .. do something else here, maybe add jobs etc.

scheduler.configure(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)

trigger 详解

date trigger

date 是最基本的一种调度，job 只会执行一次，它表示特定的时间点触发，其参数如下所示：

• run_date(datetime|str) ： job 要运行的时间，如果 run_date 为空，则默认取当前时间
• timezone(datetime.tzinfo|str) ：指定 run_date 的时区

date_example.py

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

sched = BlockingScheduler

def my_job(text):
    print(text)

# job 将在 2009 年 11 月 6 日 16:30:05 运行
sched.add_job(my_job, "date", run_date=datetime(2009, 11, 6, 16, 30, 5), args=["text"])
# 另一种写法
sched.add_job(my_job, "date", run_date="2009-11-06 16:30:05", args=["text"])

sched.start()

interval trigger

interval 表示周期性触发触发，其参数如下

• weeks(int) ：间隔礼拜数
• days(int) ：间隔天数
• hours(int) ：间隔小时数
• minutes(int) ：间隔分钟数
• seconds(int) ：间隔秒数
• start_date(datetime|str) ：周期执行的起始时间点
• end_date(datetime|str) ：最后 可能 触发时间
• timezone(datetime.tzinfo|str) ：计算 date/time 类型时需要使用的时区
• jitter(int|None) ：最多提前或延后执行 job 的 偏振 秒数

如果 start_date 为空，则默认是 datetime.now() + interval 作为起始时间。
如果 start_date 是过去的时间，trigger 不会追溯触发多次，而是根据过去的起始时间计算从当前时间开始下一次的运行时间。

interval_example.py

from datetime import datetime

from apscheduler.schedulers.blocking import BlockingScheduler


def job_function():
    print("Hello World")

sched = BlockingScheduler()

# job_function 每两个小时执行一次，同时添加了 jitter 可以增加随机性
# 防止如多个服务器在同一时间运行某个 job 时会非常有用
sched.add_job(job_function, 'interval', hours=2, jitter=120, start_date="2010-10-10 09:30:00", end_date="2014-06-15 11:00:00")

sched.start()

cron trigger

cron 提供了和 Linux crontab 格式兼容的触发器，是功能最为强大的触发器，其参数如下所示：

• year(int|str) - 4 位年份
• month(int|str) - 2 位月份(1-12)
• day(int|str) - 一个月内的第几天(1-31)
• week(int|str) - ISO 礼拜数(1-53)
• day_of_week(int|str) - 一周内的第几天(0-6 或者 mon, tue, wed, thu, fri, sat, sun)
• hour(int|str) - 小时(0-23)
• minute(int|str) - 分钟(0-59)
• second(int|str) - 秒(0-59)
• start_date(datetime|str) - 最早可能触发的时间(date/time)，含该时间点
• end_date(datetime|str) - 最后可能触发的时间(date/time)，含该时间点
• timezone(datetime.tzinfo|str) - 计算 date/time 时所指定的时区（默认为 scheduler 的时区）
• jitter(int|None) - 最多提前或延后执行 job 的 偏振 秒数

一周的开始时间总是周一！

对于 cron trigger 来说，它的强大在于可以在每个参数字段上指定各种不同的表达式来确定下一个执行时间，类似于 Unix 的 cron 程序。但和 crontab 表达式不同的是，你可以忽略不需要的字段，其行为如下 大于你显式指定的最小参数字段的参数默认都为 * ，而小于的则默认为最小值（week 和 day_of_week 除外）。 这是 ApScheduler 2.0 修正后的默认行为，在此之前忽略的字段始终默认为 * 。如 day=1, minute=20 等同于 year="*", month="*", day=1, week="*", day_of_week="*", day_of_week="*", hour="*", minute=20, second=0 。

下表列出了从年份到秒可以使用的表达式，可以在单个字段中使用逗号隔开多个表达式：

表达式	应用字段	描述
*	any	通配符
*/a	any	可被 a 整除的通配符
a-b	any	在 a-b 范围内的通配符
a-b/c	any	在 a-b 范围内可被 c 整除的通配符
xth y	day	表示一个月内的第 x 个礼拜的星期 y
last x	day	表示一个月内最后的星期 x 触发
last	day	表示月末当天触发
x,y,z	any	组合表达式，用于组合以上的表达式

cron trigger 使用所谓的 walk clock 时间，因此如果所选时区遵守 DST(Daylight saving time 夏令时)，那么在进入或退出夏令时时间时可能会导致意外发生。为了避免这个问题建议使用 UTC 时间，或提前预知并规划好执行的问题。

cron_example.py

from apscheduler.schedulers.blocking import BlockingScheduler


def job_function():
    print "Hello World"

sched = BlockingScheduler()

# job_function 会在 6、7、8、11、12 月的第三个周五的 00:00, 01:00, 02:00 以及 03:00 执行
sched.add_job(job_function, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')

# 可以使用装饰器模式
@sched.scheduled_job('cron', id='my_job_id', day='last sun')
def some_decorated_task():
    print("I am printed at 00:00:00 on the last Sunday of every month!")

# 或者直接使用 crontab 表达式
sched.add_job(job_function, CronTrigger.from_crontab('0 0 1-15 may-aug *'))

sched.start()

启动 scheduler

调用 start() 即可启动 scheduler 。对于非 BlockingScheduler 的 scheduler 来说，调用会立即返回，你可以继续你应用程序的初始化工作，例如为 scheduler 添加 job 。

对于 BlockingScheduler 来说，你只能等待 start() 函数返回之后才能继续初始化步骤。

一旦 scheduler 被启动，你就不可以再更改其设置

添加 job

有两种途径可以为 scheduler 添加 job ：

1. 调用 add_job() 方法
2. 使用 scheduled_job() 装饰一个函数

第一种方法是最常用的，第二种方法通过声明 job 而不修改应用程序运行时是最为方便的。add_job() 方法返回一个 apscheduler.job.Job 实例，你可以用它来在之后修改或移除 job 。

你可以 随时 调度 scheduler 里的 job 。如果添加 job 时，scheduler 尚未运行，job 会被临时地进行排列，直到 scheduler 启动之后，它的首次运行时间才会被确切地计算出来。

注意：

如果你希望使用 executor 或 job store 来序列化 job ，那么 job 必须满足以下两个条件：

1. （被调度的）目标里的可调用对象必须时全局可访问的
2. 可调用对象的任何参数都可以被序列化

重要事项

如果你调度的 job 在一个持久化的 job store 里，当你初始化你的应用程序时，你 必须 为 job 定义一个显式的 ID 并使用 replace_existing=True ，否则每次你的应用程序重启时你都会得到那个 job 的一个新副本。

提示

如果想马上运行 job ，请在添加 job 时省略 trigger 参数。

移除 job

当从 scheduler 中移除一个 job 时，它会从关联的 job store 中被移除，不再被执行。有两种途径可以移除 job ：

1. 通过 job 的 ID 以及 job store 的别名来调用 remove_job() 方法
2. 对你在 add_job() 中得到的 job 实例调用 remove() 方法

后者看起来更方便，实际上它要求你必须将调用 add_job() 得到的 Job 实例存储在某个地方。而对于通过 scheduled_job() 装饰器来调度 job 的就只能使用第一种方法。

如果一个 job 完成了调度（例如它的触发器不会再被触发），它会自动被移除。

例如：

job = scheduler.add_job(myfunc, 'interval', minutes=2)
job.remove()

相似的，可以使用显式地 job ID：

scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')
scheduler.remove_job('my_job_id')

暂停和恢复 job

通过 Job 实例或者 scheduler 本身你可以轻易地暂停和恢复 job 。当一个 job 被暂停，它的下一次运行时间将会被清空，同时不再计算之后的运行时间，直到这个 job 被恢复。

暂停一个 job ，使用以下方法：

• apscheduler.job.Job.pause()
• apscheduler.schedulers.base.BaseScheduler.pause_job()

而恢复一个 job ，则可以：

• apscheduler.job.Job.resume()
• apscheduler.schedulers.base.BaseScheduler.resume_job()

获取作业调度列表

可以使用 get_jobs 方法来获得机器上可处理的作业调度列表。方法会返回一个 Job 实例的列表，如果你仅仅对特定的 job store 中的 job 感兴趣，可以将 job store 的别名作为第二个参数。

更方便的做法时，使用 print_jobs() 来格式化输出作业列表以及它们的触发器和下一次的运行时间。

修改 job

通过 apscheduler.job.Job.modify() 或者 modify_job() 方法均可修改 job 的属性。你可以根据 id 修改该任何 Job 的属性。

例如：

job.modify(max_instances=6, name='Alternate name')

如果你想重新调度一个 job （这意味着要修改其 trigger），你可以使用 apscheduler.job.Job.reschedule() 或 reschedule_job() 方法。这些方法都会为 job 构建新的 trigger ，然后根据新的 trigger 重新计算其下一次的运行时间：

scheduler.reschedule_job('my_job_id', trigger='cron', minute='*/5')

终止 scheduler

以下方法可以终止 scheduler：

scheduler.shutdown()

默认情况下，scheduler 会终止其 job store 以及 executor ，然后等待所有目前执行的 job 完成后（自行终止）。如果你不想等待，可以这样：

scheduler.shutdown(wait=False)

这样依旧会终止 job store 和 executor ，但不会等待任何运行中的任务完成。

暂停/恢复 job 的运行

你可以用以下方法暂停被调度的 job 的运行：

scheduler.pause()

这会导致 scheduler 再被恢复之前一直处于休眠状态：

scheduler.resume()

如果没有进行过唤醒，也可以对处于暂停状态的 scheduler 执行 start 操作：

scheduler.start(paused=True)

这样可以让你有机会在那些不想要的 job 运行之前将它们排除掉。

限制作业的并发执行实例数目

默认情况下，每个 job 同时只会有一个实例在运行。这意味着如果一个 job 到达计划运行时间点时，前一个 job 尚未完成，那么这个 job 最近的一次运行计划将会 misfire（错过）。可以通过在添加 job 时指定 max_instances 关键字参数来设置具体 job 的最大实例数目，以便 scheduler 随后可以并发地执行它。

错过的作业执行以及合并操作（coalescing）

有时候 scheduler 无法在被调度的 job 的计划运行时间点去执行这个 job 。常见的原因是这个 job 是在持久化的 job store 中，恰好在其打算运行的时刻 scheduler 被关闭或重启了。这样，这个 job 就被定义为 misfired （错过）。scheduler 稍后会检查 job 每个被错过的执行时间的 misfire_grace_time 选项（可以单独给每个 job 设置或者给 scheduler 做全局设置），以此来确定这个执行操作是否要继续被触发。这可能到导致连续多次执行。

如果这个行为不符合你的实际需要，可以使用 coalescing 来回滚所有的被错过的执行操作为唯一的一个操作。如果对 job 启用了 coalescing ，那么即便 scheduler 在队列中看到这个 job 一个或多个执行计划，scheduler 都只会触发一次。

注意

如果因为进程（线程）池中没有可用的进程（线程）而导致 job 的运行被推迟了，那么 executor 会直接跳过它，因为相对于原计划的执行时间来说实在太晚了。如果在你的应用程序中出现了这种情况，你可以增加 executor 的线程（进程）的数目，或者调整 misfire_grace_time ，设置一个更高的值。

scheduler 事件

你可以为 scheduler 绑定事件监听器（event listen）。Scheduler 事件在某些情况下会被触发，而且它可能携带有关特定事件的细节信息。为 add_listener() 函数提供适当的掩码参数（mask argument）或者是将不同的常数组合到一起，可以监听特定类型的事件。可调用的 listener 可以通过 event object 作为参数而被调用。

留意文档里 events 模块中对于目前已有的事件以及其属性的特殊描述。

例如：

def my_listener(event):
  if event.exception:
    print('The job crashed :(')
  else:
    print('The job worked :)')
scheduler.add_listener(my_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)

故障排查

如果 scheduler 没有如预期般正常运行，可以尝试将 apscheduler 的 logger 的日志级别提升到 DEBUG 等级。

如果你还没有在一开始就将日志启用起来，那么你可以：

import logging

logging.basicConfig()
logging.getLogger('apscheduler').setLevel(logging.DEBUG)

这会提供 scheduler 运行时大量的有用信息。

高频问答环节

为什么 scheduler 不执行我的 job ？

导致这种情况的原因很多，最常见的两种情况是：

1. scheduler 在 uWSGI 的工作进程中运行，但是（uWSGI）并没有启用多线程
2. 运行了 BackgroundScheduler 但是已经执行到了脚本的末尾。

针对后一种情况，类似于这样的脚本是没办法正常工作的：

from apscheduler.schedulers.background import BackgroundScheduler

def myjob():
    print('hello')

scheduler = BackgroundScheduler()
scheduler.start()
scheduler.add_job(myjob, 'cron', hour=0)

可见，以上脚本在运行完 add_job() 之后就直接退出了，因此 scheduler 根本没有机会去运行其调度好的 job 。

我该如何在 uWSGI 中使用 APScheduler

uWSGI 使用了一些技巧来禁用掉 GIL 锁，但多线程的使用对于 APScheduler 的操作来说至关重要。为了修复这个问题，你需要使用 --enalbe-threads 选项来重新启用 GIL 。

我如何在一个或多个工作进程中共享独立的 job store

简短回答：不可以。

详细回答：在两个或更多的进程中共享一个持久化的 job store 会导致 scheduler 的行为不正常：如重复执行或作业丢失，等等。这是因为 APScheduler 目前没有任何进程间同步和信号量机制，因此当一个 job 被添加、修改或从 scheduler 中移除时 scheduler 无法得到通知。

变通方案：在专用的进程中来运行 scheduler，然后通过一些远程访问的途径 —— 如 RPyC、gRPC 或一个 HTTP 服务器 —— 来将其连接起来。在源码仓库中包含了一个使用 RPyC 的示例。

我如何在 web 应用中使用 APScheduler

首先请看上一小节的内容。

如果你想在 Django 中运行，可以考虑 django_apscheduler ，不过要注意，这个是第三方库而 APScheduler 的开发者不能保证其质量。

如果你想在 Flask 中使用 APScheduler ，这里也有一个非官方的插件 Flask-APScheduler 。

对于 Pyramid 用户而言，pyramid_scheduler 可能更有用。

对于其他情况，你最好还是按常理出牌，使用 BackgroundScheduler 。如果你在一个异步的 web 框架如 aiohttp 中运行，你可能想使用别的 scheduler 以便充分利用框架的异步功能。

APScheduler 有图形用户界面吗

简单来说，官方的没有，以下的第三方库有它们自己的实现：

• django_apscheduler
• apschedulerweb
• Nextdoor scheduler