Python定时任务框架:APScheduler源码剖析(一)

3,366 阅读8分钟

前言

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

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

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

前置概念

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

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

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

还有多种其他调度器:

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

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

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

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

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

QtScheduler: 适合使用 QT 的情况

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

剖析BackgroundScheduler

官方example代码如下。

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类的源码。

# 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方法源码如下。

# 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方法。

结尾

考虑字数,本文就先到这里,后面会继续剖析APScheduler。

如果文章对你有所帮助,点击「在看」支持二两,下篇文章见。