【Python】APScheduler简介

什么是 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。

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

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