使用APScheduler开启定时任务

前言

相比基于Linux的crontab定时任务模块来说，在Python中使用APScheduler创建定时任务更加方便、精确，编写方式也更符合Python风格，毕竟crontab里的各种**看得我心慌。且APScheduler也更适用于Django、Flask环境中添加动态任务。这篇文章就APScheduler 的使用做个简单总结。

从一个简单示例开始

先写个简单定时器示例，从代码中理解APScheduler的使用方法：

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.executors.pool import ProcessPoolExecutor
# 执行器配置：使用线程池运行定时任务，且最大线程数为5个
executors = {
    'default': ThreadPoolExecutor(max_workers=5)
}
# 配置一个调度器对象
scheduler = BackgroundScheduler(executors=executors)
# 给调度器添加定时任务task_func， 指定每日的3点运行一次
scheduler .add_job(task_func, "cron", hour=3)
# 开启定时任务
scheduler .start()

以上便是创建一个每日定时运行一次的定时任务的基本方式。

配置说明

使用apscheduler配置并开启定时任务的方法大致了解后，再去了解更多的配置项和对应实现的功能。

• 安装方式：pip install apscheduler

1. 调度器Scheduler

要实现定时任务，首先需要初始化一个调度器对象，例如上例中使用的调度器为BackgroundScheduler类，只需scheduler = BackgroundScheduler()便创建出一个调度器对象。

• BackgroundScheduler类

导入方式：from apscheduler.schedulers.background import BackgroundScheduler

适用于在框架环境中使用，如Django、Flask中，该调度器实例在调用.start()方法后不会发生阻塞，而是如其名一样在后台运行，不影响框架中其他代码的执行。

• BlockingScheduler类

from apscheduler.schedulers.blocking import BlockingScheduler

适用于作为独立进程时使用，在自己定义的脚本程序中使用BlockingScheduler类更加方便，该调度器实例在调用.start()方法后会阻塞，等待下一个定时时间点的到来再继续执行。

2. 执行器executors

在示例中使用的executor是线程池ThreadPoolExecutor的方式，对应着，当然也可以使用进程池ProcessPoolExecutor的方式。
其内部调用的本身是Python内置的concurrent.futures模块下的ThreadPoolExecutor及ProcessPoolExecutor类。

• ThreadPoolExecutor线程池
  以多线程的方式运行定时任务，一般情况使用该方案即可。
• ProcessPoolExecutor进程池
  以多进程的方式运行定时任务，当任务为CPU密集型时，应考虑选择该方案，当然，该方案会更加消耗主机资源。

使用方法如示例所示，以字典方式定义执行器，传入最大线程或进程数后，在调度器实例中指定，或使用.add_executor()添加即可。

  executors = {
      'default': ThreadPoolExecutor(20)
  }
  scheduler = BackgroundScheduler(executors=executors)
# 也可使用下面的方式添加执行器
# scheduler.add_executor()

3. 触发器Trigger

在使用sched.add_job()方法给调度器添加任务时，需要传入定时启动的方式和规定的运行时间。
以下是add_job方法源码中定义可传入的参数：

def add_job(self, func, trigger=None, args=None, kwargs=None, id=None, name=None,
                misfire_grace_time=undefined, coalesce=undefined, max_instances=undefined,
                next_run_time=undefined, jobstore='default', executor='default',
                replace_existing=False, **trigger_args):

第一个参数func为需要定时执行的任务函数名；
第三、四个参数为定时任务func的参数，如原func函数有参数，以列表或字典的形式传入即可。
第二个trigger即为触发器，指定执行任务的时机，常见的trigger参数有以下3个，具体用法释义在每个代码注释中包含：

• 1）date 在特定的时间日期执行

 from datetime import date

# 在2019年11月6日00:00:00执行
scheduler.add_job(my_job1, 'date', run_date=date(2009, 11, 6))

# 在2019年11月6日16:30:05执行，两种写法均可
scheduler.add_job(my_job2, 'date', run_date=datetime(2009, 11, 6, 16, 30, 5))
scheduler.add_job(my_job3, 'date', run_date='2009-11-06 16:30:05')

# 立即执行，不指定具体时间时会立即执行任务
scheduler.add_job(my_job4, 'date')  
scheduler.start()

• 2）interval 经过指定的时间间隔执行

from datetime import datetime

# 每两小时执行一次
scheduler.add_job(job_function, 'interval', hours=2)

# 在2019年10月10日09:30:00 到2020年6月15日的时间内，每两小时执行一次
scheduler.add_job(job_function, 'interval', hours=2, start_date='2019-10-10 09:30:00', end_date='2020-06-15 11:00:00')

interval可使用的时间参数如下：

weeks (int) – 等待的周数
days (int) – 等待的天数
hours (int) – 等待的小时数

minutes (int) – 等待的分钟数
seconds (int) – 等待的秒数
start_date (datetime|str) – 区间计算的起点时间
end_date (datetime|str) – 区间计算的终止时间
timezone (datetime.tzinfo|str) – 用于日期/时间计算的时区

• 3）cron按指定的周期执行
  在最初的示例中使用的便是该类触发器，按指定的周期循环运行。

# 在每天的3:00点运行
scheduler.add_job(task_func, "cron", hour=3)

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

# 在2019年5月30日前的周一到周五的5:30执行
scheduler.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2019-05-30')

cron可使用的时间参数如下：

year (int|str) – 年（4位数的年份）
month (int|str) – 月（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) – 最早可能触发的日期/时间（包括在内）
end_date (datetime|str) – 最晚可能触发的日期/时间（包括在内）
timezone (datetime.tzinfo|str) – 用于日期/时间计算的时区 (默认为调度程序的时区)

触发器的参数看起来内容太多，但其实都不用记住，只需在明确需求后去上述参数中查找，组合并设置自己想指定的定时方案即可。

4. 任务储存器

除以上3个核心功能外，定时任务还可以指定持久储存方案。
默认的储存器使用的是使用内存，当任务意外中断，重启后定义的任务也会重新被添加到调度器，简单而高效；
但当作业中断需要从中断中恢复任务，则在配置时应该使用数据库来作为任务储存器，具体使用什么数据库，可以自行选择SQLite、MongoDB、Redis。
定义方法如下，这里不做过多介绍，如需使用，再查找相关文档：

from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
jobstores = {
    'mongo': {'type': 'mongodb'},
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
scheduler = BackgroundScheduler(executors=executors， jobstores=jobstores)

以上。

最后，如果需要创建的定时任务太多，应该了解学习下RabbitMQ异步队列的使用。