Apscheduler：定时任务框架【提供了基于日期、固定时间间隔以及crontab类型的任务，并且可以持久化任务；基于这些功能，我们可以很方便的实现一个python定时任务系统】

我会带着你远行 2023-10-11 22:38 4阅读 0赞

### 一、概述 ###

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

apscheduler包含了四种组件，分别为：

#### 1） triggers ####

触发器包含调度逻辑，每个作业都有自己的触发器，用于确定下一次运行作业的时间，除了初始配置之外，触发器是完全无状态的

触发器主要有如下三种，分别为cron、interval和date

① cron定时调度：某一时刻执行，类似于linux的crotab

\* year (int|str) – 年，4位数字

\* month (int|str) – 月 (范围1-12)

\* day (int|str) – 日 (范围1-31)

\* week (int|str) – 周 (范围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) – 最晚结束时间(包含)

相关参数的取值格式如下：

![format_png][]

② interval：间隔调度，即每个多久执行一次

\* weeks (int) – 间隔几周

\* days (int) – 间隔几天

\* hours (int) – 间隔几小时

\* minutes (int) – 间隔几分钟

\* seconds (int) – 间隔多少秒

\* start\_date (datetime|str) – 开始日期

\* end\_date (datetime|str) – 结束日期

③ date：定时调度，作业只会被执行一次

![9b0278904e835afcc26c89b771dcd53d.png][]

from apscheduler.schedulers.background import BackgroundScheduler
    
    sche = BackgroundScheduler()
    
    # cron: 类似于linux的crontab，可以设置非常明细的定时任务
    sche.add_job(
        id="0001", name="jobname",
        func=funcname, trigger="cron",
        day="1", hour="12", minute="30"
    )
    
    # interval：固定间隔重复执行
    sche.add_job(
        id="0002", name="jobname",
        func=funcname, trigger="cron",
        seconds="30"
    )

#### 2） job stores ####

作业存储包含预定的作业，默认作业存储只是将作业保存在内存中，但其他作业将它们存储在各种数据库中；作业的数据在保存到持久作业存储时被序列化，并在从它加载回来时反序列化；作业存储（默认情况除外）不会将作业数据保存在内存中，而是充当在后端保存、加载、更新和搜索作业的中间人；作业存储绝不能在调度程序之间共享

作业存储器的选择有两种：一是内存，也是默认的配置；二是数据库

具体选哪一种看我们的应用程序在崩溃时是否重启整个应用程序，如果重启整个应用程序，那么作业会被重新添加到调度器中，此时简单的选取内存作为作业存储器即简单又高效；但是，当调度器重启或应用程序崩溃时您需要您的作业从中断时恢复正常运行，那么通常我们选择将作业存储在数据库中，使用哪种数据库通常取决于为在您的编程环境中使用了什么数据库；我们可以自由选择，PostgreSQL 是推荐的选择，因为它具有强大的数据完整性保护

APScheduler 可以配合多种不同的作业存储后端一起使用，目前支持以下的作业存储后端：

*  1、内存 Memory
 *  2、SQLAlchemy (任何 SQLAlchemy 支持的关系型数据库)
 *  3、MongoDB
 *  4、Redis
 *  5、RethinkDB
 *  6、ZooKeeper

#### 3） executors ####

Executors负责处理作业的运行；他们通常通过将作业中指定的可调用对象提交给线程或进程池来完成此操作；作业完成后，执行程序通知调度程序，然后调度程序发出适当的事件；

执行器的选择也取决于应用场景，通常默认的 ThreadPoolExecutor 已经足够好；如果作业负载涉及CPU 密集型操作，那么应该考虑使用 ProcessPoolExecutor，甚至可以同时使用这两种执行器，将ProcessPoolExecutor 行器添加为二级执行器

4） schedulers

调度程序是将其余部分绑定在一起的东西；您的应用程序中通常只有一个调度程序在运行；应用程序开发人员通常不直接处理作业存储、执行程序或触发器；相反，调度程序提供了适当的接口来处理所有这些；作业存储和执行器的配置是通过调度程序完成的，添加、修改和删除作业也是如此

各类调度器适用场景如下：

*  1、BlockingScheduler：适用于调度程序是进程中唯一运行的进程，调用start函数会阻塞当前线程，不能立即返回
 *  2、BackgroundScheduler：适用于调度程序在应用程序的后台运行，调用start后主线程不会阻塞
 *  3、AsyncIOScheduler：适用于使用了asyncio模块的应用程序
 *  4、GeventScheduler：适用于使用gevent模块的应用程序
 *  5、TwistedScheduler：适用于构建Twisted的应用程序
 *  6、QtScheduler：适用于构建Qt的应用程序

### 二、使用步骤 ###

#### 1）创建调度器 ####

from apscheduler.schedulers.blocking import BlockingScheduler
    from apscheduler.schedulers.background import BackgroundScheduler
    from apscheduler.schedulers.asyncio import AsyncIOScheduler
    from apscheduler.schedulers.gevent import GeventScheduler
    from apscheduler.schedulers.twisted import TwistedScheduler
    from apscheduler.schedulers.qt import QtScheduler
    
    bgsched = BackgroundScheduler()

#### 2）添加job ####

有两种途径可以为 scheduler 添加 job ：

① 调用 add\_job() 方法

② 使用 scheduled\_job() 装饰一个函数

第一种方法是最常用的，第二种方法通过声明 job 而不修改应用程序运行时是最为方便的；add\_job() 方法返回一个 apscheduler.job.Job 实例，我们可以用它来在之后修改或移除 job；我们可以随时调度 scheduler 里的 job，如果添加 job 时，scheduler 尚未运行，job 会被临时地进行排列，直到 scheduler 启动之后，它的首次运行时间才会被确切地计算出来

注意：

① 如果你调度的 job 在一个持久化的 job store 里，当你初始化你的应用程序时，你 必须 为 job 定义一个显式的 ID 并使用 replace\_existing=True ，否则每次你的应用程序重启时你都会得到那个 job 的一个新副本

② 如果想马上运行 job ，请在添加 job 时省略 trigger 参数

③ 如果我们的执行器或任务储存器是会序列化任务的，那么这些任务就必须符合：1-回调函数必须全局可用；2-回调函数参数必须也是可以被序列化的

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):
    
    def my_func():
        import time
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    
    myjob = bgsched.add_job(
        func=my_func, trigger='interval', id='00001', name='First job',
        jobstore='default', executor='default', replace_existing=True
    )

#### 3）移除job ####

当从 scheduler 中移除一个 job 时，它会从关联的 job store 中被移除，不再被执行；有两种途径可以移除 job：

① 通过 job 的 ID 以及 job store 的别名来调用 remove\_job() 方法

② 对你在 add\_job() 中得到的 job 实例调用 remove() 方法

后者看起来更方便，实际上它要求你必须将调用 add\_job() 得到的 Job 实例存储在某个地方；而对于通过 scheduled\_job() 装饰器来调度 job 的就只能使用第一种方法；

如果一个 job 完成了调度（例如它的触发器不会再被触发），它会自动被移除

# 移除job任务
    myjob.remove()
    bgsched.remove_job(job_id='00001') # jobstore=None

#### 4）暂停和恢复job ####

# 暂停job
    # apscheduler.job.Job.pause()
    # apscheduler.schedulers.base.BaseScheduler.pause_job()
    myjob.pause()
    
    # 恢复job
    # apscheduler.job.Job.resume()
    # apscheduler.schedulers.base.BaseScheduler.resume_job()
    myjob.resume()

#### 5）获取作业调度列表 ####

通过get\_jobs方法获取调度列表：bgsched.get\_jobs()

#### 6）修改job ####

# 修改job相关信息，id不能被修改
    myjob.modify(name='modity_name')

#### 7）终止调度 ####

默认情况下，scheduler 会终止其 job store 以及 executor，然后等待所有目前执行的 job 完成后自行终止，如果不想等待可以设置wait为False

# 终止调度
    bgsched.shutdown(wait=False)

#### 8）启动调度 ####

from apscheduler.schedulers.blocking import BlockingScheduler
    
    bgsched = BlockingScheduler()
    
    def my_func():
     import time
     print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    
    myjob = bgsched.add_job(
     func=my_func, trigger='interval', seconds=2, id='00001', name='First job',
     jobstore='default', executor='default', replace_existing=True
    )
    
    # 启动调度
    bgsched.start()

注意：有些timezone时区可能会有夏令时的问题；这个可能导致令时切换时，任务不执行或任务执行两次；避免这个问题，可以使用UTC时间，或提前预知并规划好执行的问题

from pytz import utc
    sched = BlockingScheduler(timezone=utc)

--------------------

[定时任务框架apscheduler - 知乎][apscheduler -]

[Python定时任务框架APScheduler详解\_Yunlord的博客-CSDN博客][Python_APScheduler_Yunlord_-CSDN]

[APScheduler详解\_高速公鹿P的博客-CSDN博客][APScheduler_P_-CSDN]

[format_png]: https://img-blog.csdnimg.cn/img_convert/00b249693f6574e1f55d65744949974f.webp?x-oss-process=image/format,png
[9b0278904e835afcc26c89b771dcd53d.png]: https://img-blog.csdnimg.cn/img_convert/9b0278904e835afcc26c89b771dcd53d.png
[apscheduler -]: https://zhuanlan.zhihu.com/p/491679794
[Python_APScheduler_Yunlord_-CSDN]: https://blog.csdn.net/kobepaul123/article/details/123616575
[APScheduler_P_-CSDN]: https://blog.csdn.net/qq_40145301/article/details/124689610