黑马程序员技术交流社区

标题: 【上海校区】python定时任务框架:APScheduler源码剖析 [打印本页]

作者: 陈泽 时间: 2020-1-9 10:25
标题: 【上海校区】python定时任务框架:APScheduler源码剖析
本帖最后由陈泽于 2020-1-9 10:31 编辑

前言

APScheduler是Python中知名的定时任务框架，可以很方面的满足定时执行或周期性执行程序任务等需求，类似于Linux上的crontab，但比crontab要更加强大，该框架不仅可以添加、删除定时任务，还提供多种持久化任务的功能。

APScheduler弱分布式的框架，因为每个任务对象都存储在当前节点中，只能通过人肉的形式实现分布式，如利用Redis来做。

第一次接触APScheduler会发它有很多概念，我当年第一次接触时就是因为概念太多，直接用crontab多舒服，但现在公司项目很多都基于APScheduler实现，所以来简单扒一扒的它的源码。

前置概念

用最简单的语言提示一下APScheduler中的关键概念。

Job: 任务对象，就是你要执行的任务
JobStores: 任务存储方式，默认是存储在内存中，还可以支持redis、mongodb等
Executor: 执行器，就是执行任务的东西
Trigger: 触发器，到达某个条件触发相应的调用逻辑
Scheduler: 调度器，将上面几个部分连接起来的东西

APScheduler提供多个Scheduler，不同Scheduler适用于不同的情景，目前我最常见的就是BackgroundScheduler后台调度器，该调度器适合要求在后台运行程序的调度。

还有多种其他调度器：

BlockingScheduler：适合于只在进程中运行单个任务的情况，通常在调度器是你唯一要运行的东西时使用。

AsyncIOScheduler：适合于使用 asyncio 框架的情况

GeventScheduler: 适合于使用 gevent 框架的情况

TornadoScheduler: 适合于使用 Tornado 框架的应用

TwistedScheduler: 适合使用 Twisted 框架的应用

QtScheduler: 适合使用 QT 的情况

本文只剖析 BackgroundScheduler 相关的逻辑，先简单看看官方example，然后以此为入口逐层剖析。

剖析BackgroundScheduler

官方example代码如下

[Python] 纯文本查看 复制代码

from datetime import datetime

import time

import os

from apscheduler.schedulers.background import BackgroundScheduler



def tick():

    print('Tick! The time is: %s' % datetime.now())



if __name__ == '__main__':

    scheduler = BackgroundScheduler()

    scheduler.add_job(tick, 'interval', seconds=3) # 添加一个任务，3秒后运行

    scheduler.start()

    print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))



    try:

        # 这是在这里模拟应用程序活动（使主线程保持活动状态）。

        while True:

            time.sleep(2)

    except (KeyboardInterrupt, SystemExit):

        # 关闭调度器

        scheduler.shutdown()

上述代码非常简单，先通过BackgroundScheduler方法实例化一个调度器，然后调用add_job方法，将需要执行的任务添加到JobStores中，默认就是存到内存中，更具体点，就是存到一个dict中，最后通过start方法启动调度器，APScheduler就会每隔3秒，触发名为interval的触发器，从而让调度器调度默认的执行器执行tick方法中的逻辑。

当程序全部执行完后，调用shutdown方法关闭调度器。

BackgroundScheduler其实是基于线程形式构成的，而线程就有守护线程的概念，如果启动了守护线程模式，调度器不一定要关闭。

先看一下BackgroundScheduler类的源码。

[Python] 纯文本查看 复制代码

# apscheduler/schedulers/background.py



class BackgroundScheduler(BlockingScheduler):



    _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')

        # 设置为守护线程，Python主线程运行完后，直接结束不会理会守护线程的情况，

        # 如果是非守护线程，Python主线程会在运行完后，等待其他非守护线程运行完后，再结束

        self._thread.daemon = self._daemon # daemon 是否为守护线程

        self._thread.start() # 启动线程



    def shutdown(self, *args, **kwargs):

        super().shutdown(*args, **kwargs)

        self._thread.join()

        del self._thread

上述代码中，给出了详细的注释，简单解释一下。

_configure方法主要用于参数设置，这里定义了self._daemon 这个参数，然后通过super方法调用父类的_configure方法。

start方法就是其启动方法，逻辑也非常简单，创建了线程事件Event，线程事件是一种线程同步机制，你扒开看其源码，会发现线程事件是基于条件锁来实现的，线程事件提供了set()、wait()、clear()这3个主要方法。

set()方法会将事件标志状态设置为true。
clear()方法将事件标志状态设置为false。
wait()方法会阻塞线程，直到事件标志状态为true。

创建了线程事件后，调用了其父类的start()方法，该方法才是真正的启动方法，暂时放放，启动完后，通过Thread方法创建一个线程，线程的目标函数为self._main_loop，它是调度器的主训练，调度器不关闭，就会一直执行主循环中的逻辑，从而实现APScheduler各种功能，是非常重要方法，同样，暂时放放。创建完后，启动线程就ok了。

线程创建完后，定义线程的daemon，如果daemon为True，则表示当前线程为守护线程，反之为非守护线程。

简单提一下，如果线程为守护线程，那么Python主线程逻辑执行完后，会直接退出，不会理会守护线程，如果为非守护线程，Python主线程执行完后，要等其他所有非守护线程都执行完才会退出。

shutdown方法先调用父类的shutdown方法，然后调用join方法，最后将线程对象直接del删除。

BackgroundScheduler类的代码看完了，回看一开始的example代码，通过BackgroundScheduler实例化调度器后，接着调用的是add_job方法，向add_job方法中添加了3个参数，分别是想要定时执行的tick方法，触发器trigger的名称，叫interval，而这个触发器的参数为seconds=3。

是否可以将触发器trigger的名称改成任意字符呢？这是不可以的，APScheduler在这里其实使用了Python中的entry point技巧，如果你经过过做个Python包并将其打包上传到PYPI的过程，你对entry point应该有印象。其实entry point不止可能永远打包，还可以用于模块化插件体系结构，这个内容较多，放到后面再聊。

简单而言，add_job()方法要传入相应触发器名称，interval会对应到apscheduler.triggers.interval.IntervalTrigger类上，seconds参数就是该类的参数。

剖析add_job方法

add_job方法源码如下。

[Python] 纯文本查看 复制代码

# apscheduler/schedulers/base.py/BaseScheduler



    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):

        job_kwargs = {

            'trigger': self._create_trigger(trigger, trigger_args),

            'executor': executor,

            'func': func,

            'args': tuple(args) if args is not None else (),

            'kwargs': dict(kwargs) if kwargs is not None else {},

            'id': id,

            'name': name,

            'misfire_grace_time': misfire_grace_time,

            'coalesce': coalesce,

            'max_instances': max_instances,

            'next_run_time': next_run_time

        }

        # 过滤

        job_kwargs = dict((key, value) for key, value in six.iteritems(job_kwargs) if

                          value is not undefined)

        # 实例化具体的任务对象

        job = Job(self, **job_kwargs)



        # Don't really add jobs to job stores before the scheduler is up and running

        with self._jobstores_lock:

            if self.state == STATE_STOPPED:

                self._pending_jobs.append((job, jobstore, replace_existing))

                self._logger.info('Adding job tentatively -- it will be properly scheduled when '

                                  'the scheduler starts')

            else:

                self._real_add_job(job, jobstore, replace_existing)



        return job

add_job方法代码不多，一开始，创建了job_kwargs字典，其中含有触发器、执行器等，简单理一理。

trigger触发器，通过self._create_trigger()方法创建，该方法需要两个参数，代码中的trigger其实就是interval字符串，trigger_args则为对应的参数。
exectuor执行器目前为default，这个后面再聊。
func回调方法，就是我们自己真正希望被执行的逻辑，触发器会触发调度器，调度器会调用执行器去执行的具体逻辑。
misfire_grace_time：其注释解释为「指定运行时间后几秒仍运行该任务运行」，阅读相关文档才可以理解，比如一个任务，原本12:00运行，但12:00由于某些原因没有被调度，现在12:30分了，此时调度时会判断当前时间与预调度时间的差值，如果misfire_grace_time设置为20，则不会调度执行这个此前调度失败的任务，如果misfire_grace_time设置为60，则会调度。
coalesce：如果某个任务因为某些原因没有实际运行，从而造成了任务堆积，比如堆积了10个相同的人，coalesce为True，则只执行最后一层，如果coalesce为False，则尝试连续执行10次。
max_instances：通过任务同一时间最多可以有几个实例在运行
next_run_time：任务下次运行时间

接着做了一个过滤，然后将参数传入Job类，完成任务对象的实例化。
随后的逻辑比较简单，先判断是否可以拿到self._jobstores_lock锁，它其实是一个可重入锁，Python中，可重入锁的实现基于普通互斥锁，只是多了一个变量用于计数，每加一次锁，该变量加一，每解一次锁该变量减一，只有在该变量为0时，才真正去释放互斥锁。
获取到锁后，先判断当前调度器的状态，如果是STATE_STOPPED(停止状态)则将任务添加到_pending_jobs待定列表中，如果不是停止状态，则调用_real_add_job方法，随后返回job对象。
其实_real_add_job方法才是真正的将任务对象job添加到指定存储后端的方法。
当任务对象添加到指定存储后端后(默认直接存到内存中)，调度器就会去取来执行。
回到example代码中，执行完调度器的add_job方法后，紧接着便执行调度器的start方法。

欢迎光临黑马程序员技术交流社区 (http://bbs.itheima.com/)

黑马程序员IT技术论坛 X3.2