Python 轻量级定时任务调度：APScheduler

news2024/9/22 4:14:13

简述

APscheduler (Advanced Python Scheduler)，作用为按指定的时间规则执行指定的作业。提供了基于日期date、固定时间间隔interval 、以及类似于Linux上的定时任务crontab类型的定时任务。该框架不仅可以添加、删除定时任务，还可以将任务存储到数据库中，实现任务的持久化。

pip install apscheduler

APScheduler 有四种组件及相关说明：
1） triggers（触发器）：触发器包含调度逻辑，每一个作业有它自己的触发器，用于决定接下来哪一个作业会运行，除了他们自己初始化配置外，触发器完全是无状态的。
2）job stores（作业存储）：用来存储被调度的作业，默认的作业存储器是简单地把作业任务保存在内存中，其它作业存储器可以将任务作业保存到各种数据库中，支持MongoDB、Redis、SQLAlchemy 存储方式。当对作业任务进行持久化存储的时候，作业的数据将被序列化，重新读取作业时在反序列化。
3） executors（执行器）：执行器用来执行定时任务，只是将需要执行的任务放在新的线程或者线程池中运行。当作业任务完成时，执行器将会通知调度器。对于执行器，默认情况下选择ThreadPoolExecutor就可以了，但是如果涉及到一下特殊任务如比较消耗CPU的任务则可以选择ProcessPoolExecutor，当然根据根据实际需求可以同时使用两种执行器。
4） schedulers（调度器）：调度器是将其它部分联系在一起，一般在应用程序中只有一个调度器，应用开发者不会直接操作触发器、任务存储以及执行器，相反调度器提供了处理的接口。通过调度器完成任务的存储以及执行器的配置操作，如可以添加。修改、移除任务作业。

APScheduler提供了多种调度器，可以根据具体需求来选择合适的调度器，常用的调度器有：

BlockingScheduler：适合于只在进程中运行单个任务的情况，通常在调度器是你唯一要运行的东西时使用。
BackgroundScheduler: 适合于要求任何在程序后台运行的情况，当希望调度器在应用后台执行时使用。
AsyncIOScheduler：适合于使用 asyncio 框架的情况
GeventScheduler: 适合于使用gevent框架的情况
TornadoScheduler: 适合于使用 Tornado 框架的应用
TwistedScheduler: 适合使用Twisted框架的应用
QtScheduler: 适合使用QT的情况

例子

定时执行函数test_job,每隔5秒钟执行一次

from apscheduler.schedulers.blocking import BlockingScheduler
import time
def test_job():
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time())))

test_job()

scheduler = BlockingScheduler()
scheduler.add_job(test_job,'interval',seconds=5,id='test_job')
scheduler.start()

每周星期一到星期五，下午13：00-19:00每分每4秒执行一次

import datetime
import time
from apscheduler.schedulers.blocking import BlockingScheduler
def job_function():
   print("Hello World" + " " + str(datetime.datetime.now()))
if __name__ == '__main__':
    print('start to do it')
    sched = BlockingScheduler()
    sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour='13-19', minute="*", second="*/4") # 每4秒执行一次
    sched.start()

cron 触发器

详细说明

Job 作业

Job 是 APScheduler 最小执行单位。创建 Job 时指定执行的函数，函数中所需参数，Job 执行时的一些设置信息。

id：指定作业的唯一ID
name：指定作业的名字
trigger：apscheduler定义的触发器，用于确定Job的执行时间，根据设置的trigger规则，计算得到下次执行此job的时间， 满足时将会执行
executor：apscheduler定义的执行器，job创建时设置执行器的名字，根据字符串你名字到scheduler获取到执行此
job的 执行器，执行job指定的函数
max_instances：执行此job的最大实例数，executor执行job时，根据job的id来计算执行次数，根据设置的最大实例数
来确定是否可执行
next_run_time：Job下次的执行时间，创建Job时可以指定一个时间[datetime],不指定的话则默认根据trigger获取触
发时间
misfire_grace_time：Job的延迟执行时间，例如Job的计划执行时间是21:00:00，但因服务重启或其他原因导致 21:00:31才执行，如果设置此key为40,则该job会继续执行，否则将会丢弃此job
coalesce：Job是否合并执行，是一个bool值。例如scheduler停止20s后重启启动，而job的触发器设置为5s执行一次，因此此job错过了4个执行时间，如果设置为是，则会合并到一次执行，否则会逐个执行
func：Job执行的函数
args：Job执行函数需要的位置参数
kwargs：Job执行函数需要的关键字参数

Trigger 触发器

包含调度逻辑，每一个作业有它自己的触发器，用于决定接下来哪一个作业会运行。除了它们自己初始配置以外，触发器完全是无状态的【说人话就是，这个调度逻辑只能创建时设置，创建好后无法修改调度逻辑】。

APScheduler 有三种内建的 trigger:

date: 特定的时间点触发
interval: 固定时间间隔触发
cron: 在特定时间周期性地触发

Jobstore 作业存储

如果有保存作业状态的需求的话需要使用 Jobstore
如果你的应用在每次启动的时候都会重新创建作业，那么使用默认的作业存储器（MemoryJobStore）即可，但是如果你需要在调度器重启或者应用程序奔溃的情况下任然保留作业，你应该根据你的应用环境来选择具体的作业存储器。例如：使用Mongo或者SQLAlchemy JobStore （用于支持大多数RDBMS）。

Executor 执行器

Executor在scheduler中初始化，另外也可通过scheduler的add_executor动态添加Executor。
每个executor都会绑定一个alias，这个作为唯一标识绑定到Job，在实际执行时会根据Job绑定的executor。找到实际的执行器对象，然后根据执行器对象执行Job。

Executor的种类会根据不同的调度来选择，如果选择AsyncIO作为调度的库，那么选择AsyncIOExecutor，如果选择tornado作为调度的库，选择TornadoExecutor，如果选择启动进程作为调度，选择ThreadPoolExecutor或者ProcessPoolExecutor都可以。

Executor的选择需要根据实际的scheduler来选择不同的执行器。

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