前言

在web开发中，定时任务在很多场景中是经常用到的，比较常见的定时任务就是linux自带的crontab和celery还有今天要说的这个apscheduler了。

APScheduler基于Quartz的一个Python定时任务框架，实现了Quartz的所有功能，使用起来十分方便。提供了基于日期、固定时间间隔以及crontab类型的任务，并且可以持久化任务。基于这些功能，我们可以很方便的实现一个python定时任务系统。

安装

直接使用pip安装即可

pip install apscheduler

简介

APScheduler由四部分组成，分别是：触发器，作业存储，执行器，调度器。

触发器(trigger) 包含调度逻辑，每一个作业有它自己的触发器，用于决定接下来哪一个作业会运行。除了他们自己初始配置意外，触发器完全是无状态的。

作业存储(job store) 存储被调度的作业，默认的作业存储是简单地把作业保存在内存中，其他的作业存储是将作业保存在数据库中。一个作业的数据讲在保存在持久化作业存储时被序列化，并在加载时被反序列化。调度器不能分享同一个作业存储。

执行器(executor) 处理作业的运行，他们通常通过在作业中提交制定的可调用对象到一个线程或者进城池来进行。当作业完成时，执行器将会通知调度器。

调度器(scheduler) 是其他的组成部分。你通常在应用只有一个调度器，应用的开发者通常不会直接处理作业存储、调度器和触发器，相反，调度器提供了处理这些的合适的接口。配置作业存储和执行器可以在调度器中完成，例如添加、修改和移除作业。　

简单应用

添加任务：

import time
from apscheduler.schedulers.blocking import BlockingScheduler
 
def my_job():
    print time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
 
sched = BlockingScheduler()
sched.add_job(my_job, 'interval', seconds=5)
sched.start()

上面的例子表示每隔5s执行一次my_job函数，输出当前时间信息

上面是通过add_job()来添加任务，另外还有一种方式是通过scheduled_job()修饰器来修饰函数

import time
from apscheduler.schedulers.blocking import BlockingScheduler
 
sched = BlockingScheduler()
 
@sched.scheduled_job('interval', seconds=5)
def my_job():
    print time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
 
sched.start()

移除任务：

job = scheduler.add_job(myfunc, 'interval', minutes=2)
job.remove()
#如果有多个任务序列的话可以给每个任务设置ID号，可以根据ID号选择清除对象，且remove放到start前才有效
sched.add_job(myfunc, 'interval', minutes=2, id='4')
sched.remove_job('my_job_id')

暂停任务：

apsched.job.Job.pause()
apsched.schedulers.base.BaseScheduler.pause_job()

恢复任务：

apsched.job.Job.resume()
apsched.schedulers.base.BaseScheduler.resume_job()

查询任务：

# 根据任务设置的id查看某个定时任务详情
print(sched.get_job(job_id='4'))
# 查看实例下所有定时任务
print(sched.get_jobs())

关闭调度器：

默认情况下调度器会等待所有正在运行的作业完成后，关闭所有的调度器和作业存储。

如果你不想等待，可以将wait选项设置为False。

sched.shutdown()
sched.shutdown(wait=False)

参数解析

add_job的第二个参数是trigger，它管理着作业的调度方式。它可以为date, interval或者cron。对于不同的trigger，对应的参数也相同。

  • date: 定时调度（只会执行一次）

    # 表示2009年11月6号执行一次
sched.add_job(my_job, 'date', run_date=date(2009, 11, 6), args=['text'])

  • interval: 间隔调度（每隔多久执行）

    #表示每隔3天17时19分07秒执行一次任务
sched.add_job(my_job, 'interval',days  = 03,hours = 17,minutes = 19,seconds = 07)

  • cron：定时调度（某一定时时刻执行）

    # 例子
#表示2017年3月22日17时19分07秒执行该程序
sched.add_job(my_job, 'cron', year=2017,month = 03,day = 22,hour = 17,minute = 19,second = 07)
 
#表示任务在6,7,8,11,12月份的第三个星期五的00:00,01:00,02:00,03:00 执行该程序
sched.add_job(my_job, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
 
#表示从星期一到星期五5:30（AM）直到2014-05-30 00:00:00
sched.add_job(my_job(), 'cron', day_of_week='mon-fri', hour=5, minute=30,end_date='2014-05-30')
 
#表示每5秒执行该程序一次，相当于interval 间隔调度中seconds = 5
sched.add_job(my_job, 'cron',second = '*/5')

需要注意的问题

在add_job中，如果调用的是一个有形参的函数，不能够直接在该函数后实例传参，而是应该在add_job里的args或者kwargs里传，所以，假如你有个函数，需要传参db和q

如果这样写：

ched.add_job(my_job(db,'q'), 'date', run_date='2021-04-06 14:58:40')

你会看到一个这样的错误：

ValueError: The following arguments have not been supplied: db, q

正确写法是：

schedule.add_job(hello, 'date', args=[db, 'q'], run_date='2021-04-06 14:59:00')

至于集成到web框架中，尽情期待，有时间了会补上，最后，官方文档，各位有兴趣可以看一看

补充一则很难搞的错误

我在使用apscheduler定时任务的时候，将他挂载到了fastapi上，使用起来倒是一点毛病没有，但是，我将他储存的时候，他却给我报了一个这样的错误。

_pickle.PicklingError: Can't pickle <class 'sqlalchemy.orm.session.Session'>: it's not the same object as sqlalchemy.orm.session.Session

我迟迟没有头绪，而且根据源码，我也确定，我并没有写错什么，但是这个错误就是会出现。

在我经过了不懈的寻找之后发现….是我自己写的调度函数问题，我在函数中传了一个sqlalchemy.orm.session，然而，它的储存也正好如此，所以，在传参中，他们重复了。

这个问题，应该很难遇到，因为毕竟没人跟我一样这么搞嘛，不过呢，既然遇到了我就要记录一下，以免再卡这么久。

只要我们不传session对象就好了。