Python 定时框架 APScheduler技术原理

文章目录

• • 1.APscheduler是什么
  • 2.APScdeduler的基本概念
  • • 2.1 Job 作业
    • 2.2 Trigger 触发器
    • 2.3 Executor 执行器
    • 2.4 Jobstore 作业存储器
    • 2.5 Scheduler 调度器
    • 2.6 Scheduler 事件监听机制
  • 3.APScheduler的工作流程
  • 4.APScheduler的使用实例

1.APscheduler是什么

Python定时任务框架APScheduler，Advanced Python Scheduler (APScheduler) 是一个轻量级但功能强大的进程内任务调度器，作用为在指定的时间规则执行指定的作业（时间规则：指定的日期时间、固定时间间隔以及类似Linux系统中Crontab的方式）；并且该框架可以进行持久化配置，保证在项目重启或者崩溃恢复后仍然能够恢复之前的作业继续运行。

2.APScdeduler的基本概念

2.1 Job 作业

包含要执行的任务（函数），任务执行所需要的参数，以及调度器执行该任务所需要的一些额外的配置信息，比如什么时间点执行？任务信息保存在哪里，内存还是数据库等？使用线程执行还是进程执行？

Job的属性包括：

• id：指定作业的唯一ID
• name：指定作业的名字
• trigger：触发器，分为date,interval,cron三种触发器，主要的作用就是根据设置的时间规则就是计算出任务的下一次触发时间。
• executor：执行器，负责处理作业的运行，通常使用的是线程池或进程池。当作业完成时，执行器将会通知调度器。
• max_instances：每个job在同一时刻能够运行的最大实例数,默认情况下为1个,可以指定为更大值,这样即使上个job还没运行完同一个job又被调度的话也能够再开一个线程执行
• 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执行函数需要的关键字参数。

2.2 Trigger 触发器

Trigger绑定到Job，在scheduler调度筛选Job时，根据触发器的规则计算出Job的触发时间，然后与当前时间比较确定此Job是否会被执行，总之就是根据trigger规则计算出下一个执行时间。

目前APScheduler支持触发器：

• DateTrigger： 指定日期时间执行一次
• IntervalTrigger： 固定时间间隔执行，支持每秒、每分、每时、每天、每周
• CronTrigger： 类似Linux系统的Crontab定时任务

DateTrigger和IntervalTrigger很好理解，使用也比较简单，这里重点说一下CronTrigger触发器。

CronTrigger触发器的参数选项如下：

字段	类型	说明
year	int | string	年，4位数字
month	int | string	月 (范围1-12或者jan– dec)
day	int | string	日 (范围1-31)
week	int | string	周 (范围1-53)
day_of_week	int | string	星期几 (范围0-6 或者 mon - sun)
hour	int | string	时 (范围0-23)
minute	int | string	分 (范围0-59)
second	int | string	秒 (范围0-59)
start_date	datetime | string	最早开始日期(包含)
end_date	datetime | string	最晚结束时间(包含)
timezone	datetime.tzinfo | string	指定时区

CronTrigger可用的表达式：

表达式	参数类型	描述
*	所有	通配符。例：minutes=*即每分钟触发
* / a	所有	每隔时长a执行一次。例：minutes="* / 3" 即每隔3分钟执行一次
a - b	所有	a - b的范围内触发。例：minutes=“2-5”。即2到5分钟内每分钟执行一次
a - b / c	所有	a - b范围内，每隔时长c执行一次。
xth y	日	第几个星期几触发。x为第几个，y为星期几
last x	日	一个月中，最后一个星期的星期几触发
last	日	一个月中的最后一天触发
x, y, z	所有	组合表达式，可以组合确定值或上述表达式

注：

1.当省略时间参数时，在显式指定参数之前的参数会被设定为*，之后的参数会被设定为最小值，week 和day_of_week的最小值为*。比如，设定day=1, minute=20等同于设定year=’*’, month=’*’, day=1, week=’*’, day_of_week=’*’, hour=’*’, minute=20, second=0，即每个月的第一天，且当分钟到达20时就触发。

2.如果要实现每隔一段时间执行一次的任务，有时候使用CronTriger并不合适，例如我们设定hour=’*/15’，希望任务每隔15小时执行一次，但实际上，这个表达式的意思是0-23小时内，每隔15小时执行一次，也就是每天的0：00, 15:00会执行一次; 并不会按照预期一样0:00执行一次，15:00执行一次，第二天的6:00再执行一次…

三种触发器的使用示例：

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


def task():
    print("Hello World")


if __name__ == '__main__':

    scheduler = BlockingScheduler()

    # DateTrigger 2020-12-12 00:00:00执行一次
    scheduler.add_job(task, 'date', id='date_job', run_date=datetime(2020, 12, 12, 0, 0, 0))

    # IntervalTrigger 每隔两秒执行一次
    scheduler.add_job(task, 'interval', id='interval_job', seconds=2)

    # CronTrigger 每周一到周五的5-10小时区间内，每两个小时的半点执行 一次
    scheduler.add_job(task, 'cron', id='cron_job_01', day_of_week='mon_fri', hour='5-10/2', minute=30)

    # CronTrigger 6,7,8,11,12月的第三个周五的 01:00, 02:00, 03:00执行
    scheduler.add_job(task, 'cron', id='cron_job_02', month='6-8, 11-12', day='3rd fri', hour='1-3')

    scheduler.start()
           

2.3 Executor 执行器

负责处理作业的运行，通常使用的是ThreadPoolExecutor或ProcessPoolExecutor。当作业完成时，执行器将会通知调度器。

大多数情况下， 执行器 选择 ThreadPoolExecutor 就够用了，但如果涉及到CPU密集的作业，就可以选择ProcessPoolExecutor，以充分利用多核CPU。当然也可以同时配置使用两个执行器，将进程池 ProcessPoolExecutor 调度器作为你的第二个执行器。

2.4 Jobstore 作业存储器

保存要调度的任务，其中除了默认的作业存储是把作业保存在内存中，其他的作业存储是将作业保存在数据库中。一个作业的数据将在保存在持久化的作业存储之前，会对作业执行序列化操作，当重新读取作业时，再执行反序列化操作。

目前APScheduler支持的Jobstore：

• MemoryJobStore
• MongoDBJobStore
• RedisJobStore
• RethinkDBJobStore
• SQLAlchemyJobStore
• ZooKeeperJobStore

2.5 Scheduler 调度器

负责将上面几个组件联系在一起，一般在应用中只有一个调度器，程序开发者不会直接操作触发器、作业存储或执行器，而是利用调度器提供了处理这些合适的接口，作业存储和执行器的配置都是通过在调度器中完成的。

在我们的使用过程中，选择合适的 调度器 是根据我们的开发环境以及实际应用来决定的，根据IO模型的不同，主要有下面一些常见的调度器：

• BlockingScheduler：适合于只在进程中运行单个任务的情况
• BackgroundScheduler:适合于不运行使用其他框架时，并希望在程序后台执行的情况
• AsyncIOScheduler：适合于使用asyncio框架的情况
• GeventScheduler: 适合于使用gevent框架的情况 TornadoScheduler: 适合于使用Tornado框架的应用
• TwistedScheduler: 适合使用Twisted框架的应用 QtScheduler: 适合使用QT的情况

BlockingScheduler和BackgroundScheduler的区别：

BackgroundScheduler继承于BlockingScheduler，只不过BackgroudScheduler执行轮询的操作时单开了一个线程，因此不会阻塞主程序的运行。

# APScheduler源码
class BlockingScheduler(BaseScheduler):
    """
    A scheduler that runs in the foreground
    """
    _event = None

    def start(self, *args, **kwargs):
        self._event = Event()
        super().start(*args, **kwargs)
        self._main_loop()

    def _main_loop(self):
        wait_seconds = TIMEOUT_MAX
        while self.state != STATE_STOPPED:
            self._event.wait(wait_seconds)
            self._event.clear()
            wait_seconds = self._process_jobs()
            
class BackgroundScheduler(BlockingScheduler):
    """
    A scheduler that runs in the background using a separate thread
    """

    _thread = None

    def _configure(self, config):
        self._daemon = asbool(config.pop('daemon', True))
        super()._configure(config)

    def start(self, *args, **kwargs):
        self._event = Event()
        BaseScheduler.start(self, *args, **kwargs)
        self._thread = Thread(target=self._main_loop, name='APScheduler')
        self._thread.daemon = self._daemon
        self._thread.start()
           

调度器可以通过如下三种方式进行配置：

from pytz import utc

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 = {
    'default': ThreadPoolExecutor(20),
    'processpool': ProcessPoolExecutor(5)
}
job_defaults = {
    'coalesce': False, #任务不进行归并
    'max_instances': 3	#任务最大实例数为3
}
scheduler = BackgroundScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)
           

from apscheduler.schedulers.background import BackgroundScheduler


# The "apscheduler." prefix is hard coded
scheduler = BackgroundScheduler({
    'apscheduler.jobstores.mongo': {
         'type': 'mongodb'
    },
    'apscheduler.jobstores.default': {
        'type': 'sqlalchemy',
        'url': 'sqlite:///jobs.sqlite'
    },
    'apscheduler.executors.default': {
        'class': 'apscheduler.executors.pool:ThreadPoolExecutor',
        'max_workers': '20'
    },
    'apscheduler.executors.processpool': {
        'type': 'processpool',
        'max_workers': '5'
    },
    'apscheduler.job_defaults.coalesce': 'false',
    'apscheduler.job_defaults.max_instances': '3',
    'apscheduler.timezone': 'UTC',
})
           

from pytz import utc

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ProcessPoolExecutor


jobstores = {
    'mongo': {'type': 'mongodb'},
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
executors = {
    'default': {'type': 'threadpool', 'max_workers': 20},
    'processpool': ProcessPoolExecutor(max_workers=5)
}
job_defaults = {
    'coalesce': False,
    'max_instances': 3
}
scheduler = BackgroundScheduler()

# 添加任务

scheduler.configure(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)
           

2.6 Scheduler 事件监听机制

Event是APScheduler在进行某些操作时触发相应的事件，用户可以自定义一些函数（Listener）来监听这些事件，当触发某些Event时，做一些具体的操作（比如给用户发邮件、短信等）。

常见的比如。Job执行异常事件 EVENT_JOB_ERROR。Job执行时间错过事件EVENT_JOB_MISSED。

目前APScheduler定义的Event：

• EVENT_SCHEDULER_STARTED
• EVENT_SCHEDULER_START
• EVENT_SCHEDULER_SHUTDOWN
• EVENT_SCHEDULER_PAUSED
• EVENT_SCHEDULER_RESUMED
• EVENT_EXECUTOR_ADDED
• EVENT_EXECUTOR_REMOVED
• EVENT_JOBSTORE_ADDED
• EVENT_JOBSTORE_REMOVED
• EVENT_ALL_JOBS_REMOVED
• EVENT_JOB_ADDED
• EVENT_JOB_REMOVED
• EVENT_JOB_MODIFIED
• EVENT_JOB_EXECUTED
• EVENT_JOB_ERROR
• EVENT_JOB_MISSED
• EVENT_JOB_SUBMITTED
• EVENT_JOB_MAX_INSTANCES

3.APScheduler的工作流程

当一个任务被添加到调度器，会根据指定的时间规则，计算出该任务下一次执行的时间，保存到作业存储器当中（内存或者数据库）。在调度器的主循环中，会反复检查作业储存器当中是否有到期要执行的任务，如果有就调度执行。

在主循环中，如果不间断地轮询作业存储器，但是实际上没有要执行的任务，会造成资源的浪费。因此Apscheduler在实现上，每次遍历作业存储器的时候，会找出所有任务当中离当前时间最近的任务，计算出这个任务距离下一次执行的时间间隔。这个时候，主循环就可以睡眠，等时间到了再唤醒。

4.APScheduler的使用实例

import pymongo
import logging
import datetime
import random
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
from apscheduler.jobstores.memory import MemoryJobStore
from apscheduler.jobstores.mongodb import MongoDBJobStore

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S',
    filename='log.txt',
    filemode='a'
)

job_stores = {
    'mongodb': MongoDBJobStore(
        client=pymongo.MongoClient("mongodb://localhost:27017/"),
        database='Apscheduler',
        collection='SchedulerJobInfo'
    ),
    'default': MemoryJobStore()
}

executors = {
    'process': ProcessPoolExecutor(5),
    'default': ThreadPoolExecutor(20),
}

job_defaults = {
    'coalesce': False,  # 相同任务不进行归并
    'max_instances': 1  # 单个任务的同一时刻只能执行一个实例
}


def task(arg):
    print(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), arg)
    return 1 / 0 if random.choice([True, False]) else None


def task_remove(arg):
    scheduler.remove_job('interval_task')
    print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), arg)


def task_pause(arg):
    scheduler.pause_job('interval_task')
    print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), arg)


def task_resume(arg):
    scheduler.resume_job('interval_task')
    print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), arg)


def listen(event):
    if event.exception:
        print("任务执行出错!!!")
    else:
        print("任务成功执行...")


if __name__ == '__main__':
    scheduler = BlockingScheduler(
        jobstores=job_stores,
        executors=executors,
        job_defaults=job_defaults
    )
    scheduler.add_listener(listen, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
    scheduler._logger = logging
    # 指定时间执行一次
    scheduler.add_job(task, 'date', id='date_task', args=["一次性任务"], jobstore='mongodb',
                      run_date=datetime.datetime.now() + datetime.timedelta(seconds=10))
    # 间隔触发器：每隔5s执行一次
    scheduler.add_job(task, 'interval', id='interval_task', seconds=5, args=["循环任务"], jobstore='mongodb')
    # cron触发器：每16s执行一次
    scheduler.add_job(task, 'cron', id='cron_task', second="*/16", args=["定时任务"], jobstore='mongodb')
    # 暂停任务
    scheduler.add_job(task_pause, id='pause_task', args=['暂停间隔执行任务'], jobstore='mongodb',
                      next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=20))
    # 恢复任务
    scheduler.add_job(task_resume, id='resume_task', args=['恢复间隔执行任务'], jobstore='mongodb',
                      next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=30))
    # 删除任务
    scheduler.add_job(task_remove, id='remove_task', args=['删除间隔执行任务'], jobstore='mongodb',
                      next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=40))

    scheduler.start()
           

调度器刚启动时mongodb数据库记录：

运行结果：

任务结束后，存储器中已完成的任务会被删除：

注：

定期执行任务、暂停间隔执行任务、恢复间隔执行任务、删除间隔执行任务，都是一次性任务，执行结束后，作业存储器中相应的记录会被删除。同时，因为间隔执行任务最后被删除了，所以作业存储器中就只剩下每隔16s执行一次的定时任务了。

通过运行结果也可以看出，通过CronTrigger设置的每隔16s（second=’*/16’）执行一次的任务，只会在每分钟的0s,16s,32,48s执行，不会真的严格按照每隔16s执行一次，如果要实现严格的间隔执行，请使用IntervalTrigger。