APScheduler

安装

pip install apscheduler

apscheduler的组成

1. 触发器
   每一个作业都是它自己的触发器，用于决定接下来哪一个作业被运行。除了初始配置以外，触发器是无状态的。
2. 作业存储
   存储被调度的作业，默认的作业是简单把作业保存在内存中，其他的作业存储是将作业保存在数据库中。一个作业的数据将保存在持久化作业存储是被序列化，并在加载时被反序列化。调度器不能分享同一个作业存储。
3. 执行器
   处理作业的运行，他们通常通过在作业中提交制定的可调用对象到一个线程或者进程池来进行，当作业完成时，执行器将会通知调度器。
4. 调度器
   负责将上面几个组件联系在一起，一般在应用中只有一个调度器，我们开发一般不会直接操作触发器，作业存储和执行器，而是利用调度器提供了处理这些合适的接口，作业存储和执行器都是在调度器的配置的

常见调度器

• BlockingScheduler：适合于只在进程中运行单个任务的情况

• BackgroundScheduler: 适合于不运行使用其他框架时，并希望在程序后台执行的情况

• AsyncIOScheduler：适合于使用asyncio框架的情况

• GeventScheduler: 适合于使用gevent框架的情况

• TornadoScheduler: 适合于使用Tornado框架的应用

• TwistedScheduler: 适合使用Twisted框架的应用

• QtScheduler: 适合使用QT的情况

配置调度器

Apscheduler框架提供了许多调度器的配置方法，既可以使用配置字典，也可以直接传递配置参数给调度器使用； 同时支持先初始化调度器，添加完作业任务后，再来配置调度器等。

下面是一个小例子，使用默认调度器

from apscheduler.schedulers.blocking import BlockingScheduler

def task():
    print('this is a job')

sch = BlockingScheduler()
sch.add_job(task,'interval',seconds=5)
sch.start()

再来一个复杂点的，使用两个作业存储和两个执行器

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 = {
    'defaults':ThreadPoolExecutor(10),
    'processpool':ProcessPoolExecutor(10)
}


job_default= {
    'coalesce':False, #如果coalesce为True，下次这个job被submit给executor时，只会执行1次，也就是最后这次，如果为False，那么会执行5次（不一定，因为还有其他条件，看后面misfiregracetime的解释）
    'max_instances':3 #job运行最大实例数
}

sc = BackgroundScheduler(jobstores=jobstores,executors=executors,job_default=job_default)

启动/关闭调度器

使用 start() 方法来启动调度器，其中须注意的是 BlockingScheduler 需要在初始化之后才能执行 start() ，对于其他的调度器，调用 start() 方法都会直接返回，然后可以继续执行后面的初始化操作。同时，调度器启动之后，就不能再更改它的配置了。

在默认情况下，调度器会等所有的作业任务完成后，自动关闭所有的调度器及作业存储。若在使用过程中不想等待，可以将 wait参数选项设为 False，则表示直接关闭：

sched.shutdown(wait=False)

事件监听

可以给调度器添加事件监听器，调度器事件只有在某些情况下才会被触发，并且可以携带某些有用的信息。通过给 add_listener()传递合适的 mask参数，可以只监听几种特定的事件类型，具体类型可看源码中的 event.exception或者 event.code值来做识别判断。

from apscheduler.events import EVENT_JOB_EXECUTED,EVENT_JOB_ERROR
from apscheduler.schedulers.blocking import BlockingScheduler

def lister(event):    
    if event.code == EVENT_JOB_EXECUTED:
        print('job down')
    else:
        print('error')
        
sched = BlockingScheduler()
sched.add_listener(lister,EVENT_JOB_EXECUTED|EVENT_JOB_ERROR)

作业及作业存储

jobstore提供给scheduler一个序列化jobs的统一抽象，提供对scheduler中job的增删改查接口，根据存储backend的不同，分以下几种：

• MemoryJobStore：没有序列化，jobs就存在内存里，增删改查也都是在内存中操作
• SQLAlchemyJobStore：所有sqlalchemy支持的数据库都可以做为backend，增删改查操作转化为对应backend的sql语句
• MongoDBJobStore：用mongodb作backend
• RedisJobStore: 用redis作backend

Job是框架承接目前需要执行的工作和任务，我们可以在系统运行过程中进行动态的增加、修改、删除、查询等操作。

1.添加作业

add_job() 来添加作业，还可以使用装饰器 scheduled_job来动态装饰 Job 的实际函数

@sched.scheduled_job('interval',seconds=5)
def task():
    print('t')

2.移除作业

#多个任务可使用job id,根据job id来清楚对象，remove放在start前才有效
job = sched.add_job(task,'interval',seconds=5,id='2')
job.remove()
sched.remove_job(job_id='2')

3.暂停，恢复作业

job.pause() #暂停
job.resume() #恢复

4.获得调度作业的列表

获得调度作业的列表，可以使用 get_jobs()来完成，它会返回所有的job实例，同时也可使用 print_jobs()来输出所有格式化的作业列表。也可以利用 get_job(任务ID)获取指定任务的作业列表

print(sched.get_jobs())
print(sched.get_job(job_id='2'))

5.修改作业

除了id不能修改，其他都可以

sched.reschedule_job('2',trigger='cron',minutes='*/5')
job.modify()

作业控制

add_job()方法的第二个参数是trigger，它管理着作业任务的调度方式，它可以被设置为 data、 interval、 corn三种类别。对于不同的设置类别，对应的参数也有所不同，具体如下：
1.corn定时调度

#2020-2-14 16:13:00执行
job = sch.add_job(task,'cron',year=2020,month=2,day=14,hour=16,minute=13,second=0)
sch.start()

# 2,3,4,5,10,11月的第三个星期五的0，1，2，3点执行
job = sch.add_job(task,'cron',month='2-5,10-11',day='3rd fri',hour='0-3')
sch.start()

#每周一-周五的早上3点直到2019-03-30结束
job = sch.add_job(task,'cron',day_of_week='mon-fri',hour='3',end_date='2019-03-30')
sch.start()

##参数
"""
year  年
month  月
day  日
week  周一-周五
day_of_week 时间段周几-周几
hour 时
minutes 分
second 秒
"""

2.interval间隔调度

# interval
#每隔4天3小时执行一次
job = sch.add_job(task,'interval',days=4,hours=3)
sch.start()

#参数
'''
weeks 周
days 天
hours 时
minutes 分
seconds 秒
'''

3.data定时调度（到时间仅执行一次）

from datetime import date
import datetime
#data
# interval
#每隔4天3小时执行一次
job = sch.add_job(task,'date',run_date='2020-02-14 16:38:05',args=["text"])
sch.start()
import apscheduler
#参数
'''
run_date 日期元组 或者直接填时间
'''