什么是APScheduler

APScheduler是一个用于任务调度和定时任务管理的Python库。它提供了一个简单而灵活的方式来定义、调度和执行任务。

APScheduler的架构由以下几个核心组件组成：

调度器（Scheduler）：调度器负责管理任务的调度和执行。它可以根据预定的时间表触发任务的运行，支持不同类型的调度器（如定时、间隔、CRON表达式等）。
触发器（Trigger）：触发器定义了任务运行的时间触发条件。APScheduler提供了多种内置触发器，例如date触发器、interval触发器和cron触发器，用于定义不同类型的时间表。
作业（Job）：作业是需要调度和执行的任务。你可以定义自己的作业类，并在其中指定要执行的任务逻辑。
执行器（Executor）：执行器定义了任务的执行方式。APScheduler提供了不同的执行器，包括默认的线程执行器和进程执行器，以及基于异步框架的执行器（如Tornado、Twisted和Asyncio）。

在这里插入图片描述

如何正确选择

选择正确的调度器（scheduler）、作业存储（job store）、执行器（executor）和触发器（trigger）取决于环境以及如何使用APScheduler。根据应用程序使用的框架和技术，选择相应的调度器来与应用程序集成。下面是选择调度器的简要指南：

BlockingScheduler：当调度器是进程中唯一运行的内容时使用。
BackgroundScheduler：当没有使用下面提到的任何框架，并且希望调度器在应用程序内后台运行时使用。
AsyncIOScheduler：当应用程序使用asyncio模块时使用。
GeventScheduler：当应用程序使用gevent时使用。
TornadoScheduler：当正在构建一个Tornado应用程序时使用。
TwistedScheduler：当正在构建一个Twisted应用程序时使用。
QtScheduler：当正在构建一个Qt应用程序时使用。

其中，BlockingScheduler是最常用的调度器，适用于调度器是进程中唯一运行的情况。BackgroundScheduler适用于希望在应用程序内部以后台方式运行调度器的情况。而其他调度器如AsyncIOScheduler、GeventScheduler、TornadoScheduler、TwistedScheduler和QtScheduler适用于特定的框架和应用程序类型。

当安排一个作业时，需要选择一个触发器（trigger）。触发器确定作业运行时计算日期/时间的逻辑。APScheduler提供了三种内置的触发器类型：

date（日期）：在特定的时间点运行作业时使用。
interval（间隔）：在固定的时间间隔内运行作业时使用。
cron（定时）：在每天的特定时间点周期性地运行作业时使用。

还可以将多个触发器组合成一个触发器，它会在所有参与的触发器所约定的时间点触发，或者当任何一个触发器触发时激活。有关组合触发器的更多信息，请参阅相关文档。

添加作业/任务

有两种方法可以向调度器添加作业：

通过调用add_job()函数
通过使用@scheduled_job()装饰函数

add_job()方法是最常用的方法。@scheduled_job()方法主要是为了方便声明在应用程序运行时不会改变的作业。add_job()方法返回一个apscheduler.job.Job实例，可以使用该实例在以后修改或删除该作业。

可以随时在调度器上安排作业。如果在添加作业时调度器尚未运行，作业将被暂时安排，其第一次运行时间将在调度器启动时计算。

重要提示：如果使用序列化作业的执行器或作业存储，将对作业添加一些要求：

目标可调用函数必须是全局可访问的。
可调用函数的任何参数都必须是可序列化的。
在内置的作业存储中，只有MemoryJobStore不会序列化作业。在内置的执行器中，只有ProcessPoolExecutor会序列化作业。

快速安装与使用

安装

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple apscheduler

使用步骤

APScheduler 使用起来还算是比较简单。运行一个调度任务只需要以下三部曲。

新建一个 schedulers (调度器) 。
添加一个调度任务(job stores)。
运行调度任务。

示例代码

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

def test_tick():
    print('The time is: %s' % datetime.now())

if __name__ == '__main__':
    scheduler = BlockingScheduler()
    scheduler.add_job(test_tick, 'cron',minute='20',second='50')

    try:
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        pass

注意：任务启动超时的问题：增加misfire_grace_time参数，提高容错能力，如下代码。

scheduler.add_job(set_pv_forecast, 'cron',hour = hour, minute = minute, second=second, misfire_grace_time=120)

如何使用ctrl+break键优雅退出服务

在程序最上面的代码中，加入如下内容。

import signal
import sys

def exit_gracefully():
    sys.exit(0)

signal.signal(signal.SIGINT, exit_gracefully)
signal.signal(signal.SIGTERM, exit_gracefully)