python中定时任务apscheduler库用法详解

1、APScheduler简介

        APScheduler是Python的一个定时任务框架，用于执行周期或者定时任务，该框架不仅可以添加、删除定时任务，还可以将任务存储到数据库中，实现任务的持久化，使用起来非常方便。

        APscheduler全称Advanced Python Scheduler，作用为在指定的时间规则执行指定的作业，其是基于Quartz的一个Python定时任务框架，实现了Quartz的所有功能，使用起来十分方便。提供了基于日期、固定时间间隔以及crontab类型的任务，并且可以持久化任务。

2、APScheduler库安装

首先安装apscheduler库：

pip install apscheduler

3、APScheduler组成

1. 触发器(trigger)：包含调度逻辑，每一个作业有它自己的触发器，用于决定接下来哪一个作业会运行。除了他们自己初始配置以外，触发器完全是无状态的。
2. 作业存储(job store)：存储被调度的作业，默认的作业存储是简单地把作业保存在内存中，其他的作业存储是将作业保存在数据库中。一个作业的数据将在保存在持久化作业存储时被序列化，并在加载时被反序列化。调度器不能分享同一个作业存储。
3. 执行器(executor)：处理作业的运行，他们通常通过在作业中提交制定的可调用对象到一个线程或者进城池来进行。当作业完成时，执行器将会通知调度器。
4. 调度器(scheduler)：其他的组成部分。通常在应用只有一个调度器，应用的开发者通常不会直接处理作业存储、调度器和触发器，相反，调度器提供了处理这些的合适的接口。配置作业存储和执行器可以在调度器中完成，例如添加、修改和移除作业。

3.1 触发器(trigger)

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

APScheduler 有三种内建的 trigger:

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

简单理解：触发器就是根据你指定的触发方式，比如是按照时间间隔，还是按照 cron触发，触发条件是什么等。每个任务都有自己的触发器。

3.2 作业存储(job store)

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

        任务存储器是可以存储任务的地方，默认情况下任务保存在内存，也可将任务保存在各种数据库中。任务存储进去后，会进行序列化，然后也可以反序列化提取出来，继续执行。

3.3 执行器(executor)

        Executor在scheduler中初始化，另外也可通过scheduler的add_executor动态添加Executor。 

        每个executor都会绑定一个alias，这个作为唯一标识绑定到Job，在实际执行时会根据Job绑定的executor。找到实际的执行器对象，然后根据执行器对象执行Job。

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

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

3.4 调度器(scheduler)

        Scheduler是APScheduler的核心，所有相关组件通过其定义。scheduler启动之后，将开始按照配置的任务进行调度。除了依据所有定义Job的trigger生成的将要调度时间唤醒调度之外。当发生Job信息变更时也会触发调度。

        scheduler可根据自身的需求选择不同的组件，如果是使用AsyncIO则选择AsyncIOScheduler，使用tornado则选择TornadoScheduler。

        任务调度器是属于整个调度的总指挥官。它会合理安排作业存储器、执行器、触发器进行工作，并进行添加和删除任务等。调度器通常是只有一个的。开发人员很少直接操作触发器、存储器、执行器等。因为这些都由调度器自动来实现了。

4、使用详情

参数说明：

• 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执行函数需要的关键字参数

4.1 interval触发器

固定时间间隔触发。interval 间隔调度，参数如下：

示例代码：

from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler


def task():
    now = datetime.now()
    ts = now.strftime("%Y-%m-%d %H:%M:%S")
    print(ts)


def task2():
    now = datetime.now()
    ts = now.strftime("%Y-%m-%d %H:%M:%S")
    print(ts + ' 666!')


def task3():
    now = datetime.now()
    ts = now.strftime("%Y-%m-%d %H:%M:%S")
    print(ts + ' 888!')


def func():
    # 创建调度器BlockingScheduler()
    scheduler = BlockingScheduler()
    scheduler.add_job(task, 'interval', seconds=3, id='test_job1')
    # 添加任务，时间间隔为5秒
    scheduler.add_job(task2, 'interval', seconds=5, id='test_job2')
    # 在2022-10-27 21:50:30和2022-10-27 21:51:30之间，时间间隔为6秒
    scheduler.add_job(task3, 'interval', seconds=6, start_date='2022-10-27 21:53:00', end_date='2022-10-27 21:53:30', id ='test_job3')
    # 每小时（上下浮动20秒区间内）运行task
    # jitter振动参数，给每次触发添加一个随机浮动秒数，一般适用于多服务器，避免同时运行造成服务拥堵。
    scheduler.add_job(task, 'interval', hours=1, jitter=20, id='test_job4')
    scheduler.start()


func()

运行结果：

4.2 触发器date的使用

        date 是最基本的一种调度，作业任务只会执行一次。它表示特定的时间点触发。它的参数如下：

参数说明run_date(datetime or str)任务运行的日期或者时间timezone(datetime.tzinfo or str)指定时区

注意：run_date参数可以是date类型、datetime类型或文本类型。

示例代码：

from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler


def task():
    now = datetime.now()
    ts = now.strftime("%Y-%m-%d %H:%M:%S")
    print(ts)


def task2():
    now = datetime.now()
    ts = now.strftime("%Y-%m-%d %H:%M:%S")
    print(ts + '666!')


def func():
    # 创建调度器BlockingScheduler()
    scheduler = BlockingScheduler()
    scheduler.add_job(task, 'date', run_date=datetime(2022, 10, 27, 21, 39, 00), id='test_job1')
    scheduler.add_job(task2, 'date', run_date=datetime(2022, 10, 27, 21, 39, 50), id='test_job2')
    scheduler.start()


func()

运行结果：

4.3 cron触发器

在特定时间周期性地触发，和Linux crontab格式兼容。它是功能最强大的触发器。

cron参数：

表达式：

示例代码1：

from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler


def task():
    now = datetime.now()
    ts = now.strftime("%Y-%m-%d %H:%M:%S")
    print(ts)


def task2():
    now = datetime.now()
    ts = now.strftime("%Y-%m-%d %H:%M:%S")
    print(ts + ' 666!')


def task3():
    now = datetime.now()
    ts = now.strftime("%Y-%m-%d %H:%M:%S")
    print(ts + ' 888!')


def func():
    # 创建调度器BlockingScheduler()
    scheduler = BlockingScheduler()
    # 在每年 1-3、7-9 月份中的每个星期一、二中的 00:00, 01:00, 02:00 和 03:00 执行 task 任务
    scheduler.add_job(task, 'cron', month='1-3,7-9', day_of_week='1-2', hour='0-3', id='test_job1')
    scheduler.start()


func()

示例代码2：  【注意：day_of_week()中mon~sun对应的数字是0~6】

from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler


def task():
    now = datetime.now()
    ts = now.strftime("%Y-%m-%d %H:%M:%S")
    print(ts)


def func():
    # 创建调度器BlockingScheduler()
    scheduler = BlockingScheduler()
    # 在每个星期三中的 23:02执行 task 任务
    scheduler.add_job(task, 'cron', day_of_week='2', hour='23', minute='2')
    scheduler.start()


if __name__ == '__main__':
    func()

运行结果：

示例代码3：  【注意：day-of_week()中可以定时多天，时分秒可以是字符串也可以是数值型】

from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler


def task():
    now = datetime.now()
    ts = now.strftime("%Y-%m-%d %H:%M:%S")
    print(ts)


def func():
    # 创建调度器BlockingScheduler()
    scheduler = BlockingScheduler()
    # 在每个星期二、三中的 23:11:00执行 task 任务
    scheduler.add_job(task, 'cron', day_of_week='1-2', hour='23', minute='11', second='00')
    # 在每个星期二、三中的 23:11:03执行 task 任务
    scheduler.add_job(task, 'cron', day_of_week='1-2', hour=23, minute=11, second=3)
    # 在每个星期三、四中的 23:11:05执行 task 任务
    scheduler.add_job(task, 'cron', day_of_week='2-3', hour='23', minute='11', second='05')
    # 在每个星期三、四中的 23:11:08执行 task 任务
    scheduler.add_job(task, 'cron', day_of_week='2-3', hour=23, minute=11, second=8)
    scheduler.start()


if __name__ == '__main__':
    func()

运行结果：

参考博文：

Python定时任务框架APScheduler详解_Yunlord的博客-CSDN博客_apscheduler

python定时任务最强框架APScheduler详细教程_jspython的博客-CSDN博客_apscheduler coalesce=true