Celery源码笔记（六）定时任务

定时任务
定时任务是celery的一个重要功能，本节会对celery的定时任务进行分析。
根据之前分析可知，在celery启动的过程中，会创建并启动Beat组件，而celery的任务调度服务就在Beat组件启动的过程中被启动了。
首先，我们来看celery.apps.beat.Beat类的部分声明

class Beat(object):
    """Beat as a service."""

    # 调度服务
    Service = beat.Service
    app = None

该类中的属性Service即为调度服务，默认指向celery.beat.Service类，我们将目光转向该类的声明

class Service(object):
    """Celery periodic task service."""

    # 调度器
    scheduler_cls = PersistentScheduler

可以看到，该类中有一属性为调度器，指向了celery.beat.PersistentScheduler类，从这里我们可以分析得出，服务的调度类为PersistentScheduler类，该类又继承自同一文件中声明的Scheduler类。继承关系如下图所示：

image.png

在创建了celery.beat.Service实例后，会调用该实例的start()方法，在改方法中，定义如下

def start(self, embedded_process=False):
        info('beat: Starting...')
        debug('beat: Ticking with max interval->%s',
              humanize_seconds(self.scheduler.max_interval))

        signals.beat_init.send(sender=self)
        if embedded_process:
            signals.beat_embedded_init.send(sender=self)
            platforms.set_process_title('celery beat')

        try:
            while not self._is_shutdown.is_set():
                interval = self.scheduler.tick()
                if interval and interval > 0.0:
                    debug('beat: Waking up %s.',
                          humanize_seconds(interval, prefix='in '))
                    time.sleep(interval)
                    if self.scheduler.should_sync():
                        self.scheduler._do_sync()
        except (KeyboardInterrupt, SystemExit):
            self._is_shutdown.set()
        finally:
            self.sync()

其中可以看到，该方法创建了一个事件循环，其中启动了调度服务，而且我们可以看到在循环中有scheduler相关的方法调用，接下来，我们寻找相关的声明，来看看scheduler是怎样初始化的

def get_scheduler(self, lazy=False,
                      extension_namespace='celery.beat_schedulers'):
        filename = self.schedule_filename
        aliases = dict(
            load_extension_class_names(extension_namespace) or {})
        return symbol_by_name(self.scheduler_cls, aliases=aliases)(
            app=self.app,
            schedule_filename=filename,
            max_interval=self.max_interval,
            lazy=lazy,
        )

    @cached_property
    def scheduler(self):
        return self.get_scheduler()

这里可以看到，celery.beat.Service实例中，并没有声明scheduler属性，而是声明了一个方法并且用一个cached_property装饰器对其进行了修饰。该装饰器的作用是将方法调用的结果缓存起来，以便下次可以直接使用。

而第一次调用该方法时，将会调用get_scheduler创建一个self.scheduler_cls类型的实例，一般地，该实例的类型为PersistentScheduler。在使用中，可以将scheduler方法以调用属性的方法使用，可以理解为以单例模式使用scheduler。

回到celery.beat.Service.start()方法，我们可以看到该方法中首先调用了self.scheduler.tick()获取间隔时间，之后会sleep该时间间隔后，再通过self.scheduler.should_sync()判断是否需要进行同步，若需要则调用self.scheduler._do_sync()进行同步。

接下来，我们来分析celery.beat.Scheduler.tick()是怎样实现任务调度的，同样地，首先来看该方法的声明

def tick(self, event_t=event_t, min=min, heappop=heapq.heappop,
             heappush=heapq.heappush):
        """Run a tick - one iteration of the scheduler.

        Executes one due task per call.

        Returns:
            float: preferred delay in seconds for next call.
        """
        adjust = self.adjust
        max_interval = self.max_interval

        if (self._heap is None or
                not self.schedules_equal(self.old_schedulers, self.schedule)):
            self.old_schedulers = copy.copy(self.schedule)
            self.populate_heap()

        H = self._heap # 使用小根堆作为任务队列

        if not H:
            return max_interval

        event = H[0]
        entry = event[2]
        is_due, next_time_to_run = self.is_due(entry)
        if is_due:
            verify = heappop(H)
            if verify is event:
                next_entry = self.reserve(entry)
                self.apply_entry(entry, producer=self.producer)
                heappush(H, event_t(self._when(next_entry, next_time_to_run),
                                    event[1], next_entry))
                return 0
            else:
                heappush(H, verify)
                return min(verify[0], max_interval)
        return min(adjust(next_time_to_run) or max_interval, max_interval)

我们可以看到，这里使用了一个heap存储任务列表，每次从堆顶取出一个事件，若该事件通过校验，则对该事件进行调度并将下次执行的事件放入堆中，否则只将该事件放入堆中，最后返回下次的任务执行时间与最大间隔时间中的最小值

接下来分析celery.beat.Scheduler.should_sync()方法

def should_sync(self):
        return (
            (not self._last_sync or
               (monotonic() - self._last_sync) > self.sync_every) or
            (self.sync_every_tasks and
                self._tasks_since_sync >= self.sync_every_tasks)
        )

可以看到，该方法的实现比较简单，总结一下就是若上次同步的时间差大于最大时间差最大限制，则返回True 最后，

我们来分析celery.beat.Scheduler._do_sync()方法

class Scheduler(object):
    def _do_sync(self):
        try:
            debug('beat: Synchronizing schedule...')
            self.sync()
        finally:
            self._last_sync = monotonic()
            self._tasks_since_sync = 0

    def sync(self):
        pass

class PersistentScheduler(Scheduler):
    def sync(self):
        if self._store is not None:
            self._store.sync()

这里可以看到，该方法中调用了sync()方法，而在PersistentScheduler中重写了sync()方法，这里是使用了Python的shelve模块创建了一个存储的文件，并将内存中的数据存储至该文件中（实际上就是进行了一次持久化操作）。

image.png

简单总结一下，celery的任务调度的核心是使用一个最小堆实现优先级队列，并且每隔一段时间将调度的任务队列持久化至文件中，以防进程中断造成任务丢失的问题。

参考资料

Python内置函数property()使用实例
 红黑树 vs 最小堆
 shelve -- 用来持久化任意的Python对象

Celery源码笔记（六） 定时任务

参考资料

Celery源码笔记（六）定时任务