[源码分析] 并行分布式任务队列 Celery 之 Timer & Heartbeat

class Blueprint(bootsteps.Blueprint):
    """Worker bootstep blueprint."""

    name = 'Worker'
    default_steps = {
        'celery.worker.components:Hub', # 这里是 Hub
        'celery.worker.components:Pool',
        'celery.worker.components:Beat',
        'celery.worker.components:Timer', # 这里是 Timer
        'celery.worker.components:StateDB',
        'celery.worker.components:Consumer',
        'celery.worker.autoscale:WorkerComponent',
    }

我们首先来到 Timer Step。

从 Timer 组件的定义中可以看到，Timer 组件会根据当前worker是否使用事件循环机制来决定创建什么类型的timer。

如果使用 eventloop，则使用kombu.asynchronous.timer.Timer as _Timer，这里具体等待动作由用户自己完成。
否则使用 Pool 内部的Timer类（就是 timer_cls='celery.utils.timer2.Timer'），timer2 自己做了一个线程来做定时等待；

定义如下：

from kombu.asynchronous.timer import Timer as _Timer

class Timer(bootsteps.Step):
    """Timer bootstep."""

    def create(self, w):
        if w.use_eventloop:                        # 检查传入的Worker是否使用了use_eventloop
            # does not use dedicated timer thread.
            w.timer = _Timer(max_interval=10.0)    # 直接使用kombu的timer做定时器
        else:
            if not w.timer_cls:                     # 如果配置文件中没有配置timer_clas
                # Default Timer is set by the pool, as for example, the
                # eventlet pool needs a custom timer implementation.
                w.timer_cls = w.pool_cls.Timer      # 使用缓冲池中的Timer
            w.timer = self.instantiate(w.timer_cls,
                                       max_interval=w.timer_precision,
                                       on_error=self.on_timer_error,
                                       on_tick=self.on_timer_tick)  # 导入对应的类并实例化

起初看代码时候很奇怪，为什么要再单独定义一个 timer2？

原因推断是（因为对 Celery 的版本发展历史不清楚，所以此处不甚确定，希望有同学可以指正）：依据底层 Transport 的设计来对 Timer 做具体实现调整。

2.1 Transport

大家知道，Celery 是依赖于 Kombu，而在 Kombu 体系中，用 transport 对所有的 broker 进行了抽象，为不同的 broker 提供了一致的解决方案。通过Kombu，开发者可以根据实际需求灵活的选择或更换broker。

我们再回顾下具体 Kombu 的概念：

Connection 是 AMQP 对连接的封装；
Channel 是 AMQP 对 MQ 操作的封装；

那么两者的关系就是对 MQ 的操作（Channel）必然离不开连接（Connection），但是 Kombu 并不直接让 Channel 使用 Connection 来发送 / 接受请求，而是引入了一个新的抽象 Transport。Transport 负责具体的 MQ 的操作，也就是说 Channel 的操作都会落到 Transport 上执行；

Transport 代表真实的 MQ 连接，也是真正连接到 MQ（ redis / rabbitmq ）的实例。就是存储和发送消息的实体，用来区分底层消息队列是用 amqp、Redis 还是其它实现的。

具体 Kombu 逻辑如下图，Transport 在左下角处：

2.2 Thread-less VS Thread-based

对于 Transport，某些 rate-limit implementation（比如 RabbitMQ / Redis ）为了减少开销，采用了event-loop（底层使用了 Epoll），是 thread-less and lock-free。

而其他旧类型的 Transport 就是 Thread based，比如 Mongo。因此，

对于 Thread-less Transport
- Kombu 就采用了 kombu.asynchronous.timer.Timer as _Timer，具体等待操作是在 event-loop 中实现，就是调用者自己会做等待。
- 具体比如在 Redis Transport 之中，就有 register_with_event_loop 函数用来在 loop（就是 event-loop）中注册自己，具体如下：
- def register_with_event_loop(self, connection, loop):
  cycle = self.cycle
  cycle.on_poll_init(loop.poller)
  cycle_poll_start = cycle.on_poll_start
  add_reader = loop.add_reader
  on_readable = self.on_readable
```
    def on_poll_start():
        cycle_poll_start()
        [add_reader(fd, on_readable, fd) for fd in cycle.fds]
    loop.on_tick.add(on_poll_start)
    loop.call_repeatedly(10, cycle.maybe_restore_messages)
    loop.call_repeatedly(
        health_check_interval,
        cycle.maybe_check_subclient_health
    )
```
对于 thread-based Transport，
- 则采用了 celery.utils.timer2.Timer，timer2 自己继承了线程类，使用自己这个线程来做定时等待；
- 比如在 Mongodb transport 之中，就没有任何关于 event loop 的操作。

即，选用 timer 的哪种实现，看是否需要等待来决定，就是谁来完成 “等待” 这个动作。

翻了翻 Celery 2.4.7 的代码，发现在这个版本，确实只有 Thread-based timer，其代码涵盖了目前的 timer 2 和 kombu.asynchronous.timer.Timer 大部分功能。应该是从 3.0.2 之后，把部分代码分离到了 kombu.asynchronous.timer.Timer ，实现了 Thread-less 和 Thread-based 两个不同的实现。

具体可以参见下面源码中的注释：

- RabbitMQ/Redis: thread-less and lock-free rate-limit implementation.

    This means that rate limits pose minimal overhead when used with
    RabbitMQ/Redis or future transports using the event-loop,
    and that the rate-limit implementation is now thread-less and lock-free.

    The thread-based transports will still use the old implementation for
    now, but the plan is to use the timer also for other
    broker transports in Celery 3.1.

注意，上面的是 Timer Step，是一个启动的阶段，其目的是生成 Timer 组件给其他组件使用，并不是 Timer 功能类。

我们其次来看看 Timer 功能类在线程池 Pool 中的使用，就对应了前面 Blueprint step 之中的两种不同 cases。

分别也对应了两种应用场景（或者说是线程池实现）：

gevent 和 eventlet 使用 kombu.asynchronous.timer.Timer。
BasePool（以及其他类型线程池）使用了 timer2.Timer。

初步来分析，gevent 和 eventlet 都是用协程来模拟线程，所以本身具有Event loop，因此使用 kombu.asynchronous.timer.Timer 也算顺理成章。

3.1 gevent 和 eventlet

对于 gevent，eventlet 这种情况，使用了 class Timer(_timer.Timer) 作为 Timer 功能类。

从代码中可以看到，class Timer 扩展了 kombu.asynchronous.timer.Timer。

from kombu.asynchronous import timer as _timer

class Timer(_timer.Timer):

    def __init__(self, *args, **kwargs):
        from gevent import Greenlet, GreenletExit

        class _Greenlet(Greenlet):
            cancel = Greenlet.kill

        self._Greenlet = _Greenlet
        self._GreenletExit = GreenletExit
        super().__init__(*args, **kwargs)
        self._queue = set()

    def _enter(self, eta, priority, entry, **kwargs):
        secs = max(eta - monotonic(), 0)
        g = self._Greenlet.spawn_later(secs, entry)
        self._queue.add(g)
        g.link(self._entry_exit)
        g.entry = entry
        g.eta = eta
        g.priority = priority
        g.canceled = False
        return g

    def _entry_exit(self, g):
        try:
            g.kill()
        finally:
            self._queue.discard(g)

    def clear(self):
        queue = self._queue
        while queue:
            try:
                queue.pop().kill()
            except KeyError:
                pass

    @property
    def queue(self):
        return self._queue

3.2 BasePool

而 BasePool 采用了 timer2 . Timer 作为 Timer 功能类。

from celery.utils import timer2

class BasePool:
    """Task pool."""

    Timer = timer2.Timer

下面我们具体看看 Timer 功能类如何实现。

4.1 异步

kombu.asynchronous.timer.Timer 实现了异步Timer。

由其注释可以，kombu.asynchronous.timer.Timer 在调用者每次得到下一次entry时，会给出tuple of (wait_seconds, entry)，调用者应该进行等待相应时间。

即，kombu.Timer是调用者等待，普通timer是timer自己启动线程等待。

"""Iterate over schedule.
This iterator yields a tuple of ``(wait_seconds, entry)``,
where if entry is :const:`None` the caller should wait
for ``wait_seconds`` until it polls the schedule again.
"""

定义如下：

class Timer:
    """Async timer implementation."""

    Entry = Entry

    on_error = None

    def __init__(self, max_interval=None, on_error=None, **kwargs):
        self.max_interval = float(max_interval or DEFAULT_MAX_INTERVAL)
        self.on_error = on_error or self.on_error
        self._queue = []

4.2 调用

4.2.1 添加 timer function

用户通过 call_repeatedly 来添加 timer function。

def call_repeatedly(self, secs, fun, args=(), kwargs=None, priority=0):
    kwargs = {} if not kwargs else kwargs
    tref = self.Entry(fun, args, kwargs)

    @wraps(fun)
    def _reschedules(*args, **kwargs):
        last, now = tref._last_run, monotonic()
        lsince = (now - tref._last_run) if last else secs
        try:
            if lsince and lsince >= secs:
                tref._last_run = now
                return fun(*args, **kwargs) # 调用用户方法
        finally:
            if not tref.canceled:
                last = tref._last_run
                next = secs - (now - last) if last else secs
                self.enter_after(next, tref, priority)

    tref.fun = _reschedules
    tref._last_run = None
    return self.enter_after(secs, tref, priority)

4.2.2 调用

Timer通过apply_entry进行调用。

def apply_entry(self, entry):
    try:
        entry()
    except Exception as exc:
        if not self.handle_error(exc):
            logger.error('Error in timer: %r', exc, exc_info=True)

在获取下一次entry时，会返回等待时间。

def __iter__(self, min=min, nowfun=monotonic,
             pop=heapq.heappop, push=heapq.heappush):
    """Iterate over schedule.

    This iterator yields a tuple of ``(wait_seconds, entry)``,
    where if entry is :const:`None` the caller should wait
    for ``wait_seconds`` until it polls the schedule again.
    """
    max_interval = self.max_interval
    queue = self._queue

    while 1:
        if queue:
            eventA = queue[0]
            now, eta = nowfun(), eventA[0]

            if now < eta:
                yield min(eta - now, max_interval), None
            else:
                eventB = pop(queue)

                if eventB is eventA:
                    entry = eventA[2]
                    if not entry.canceled:
                        yield None, entry
                    continue
                else:
                    push(queue, eventB)
        else:
            yield None, None

4.3 实验

我们做实验看看 timer 功能类的使用。

4.3.1 示例代码

下面代码来自https://github.com/liuliqiang/blog_codes/tree/master/python/celery/kombu，特此感谢。

def main(arguments):
    hub = Hub()
    exchange = Exchange('asynt')
    queue = Queue('asynt', exchange, 'asynt')

    def send_message(conn):
        producer = Producer(conn)
        producer.publish('hello world', exchange=exchange, routing_key='asynt')
        print('message sent')

    def on_message(message):
        print('received: {0!r}'.format(message.body))
        message.ack()
        # hub.stop()  # <-- exit after one message

    conn = Connection('redis://localhost:6379')
    conn.register_with_event_loop(hub)

    def p_message():
        print('redis://localhost:6379')

    with Consumer(conn, [queue], on_message=on_message):
        send_message(conn)
        hub.timer.call_repeatedly(
            3, p_message
        )
        hub.run_forever()

if __name__ == '__main__':
    sys.exit(main(sys.argv[1:]))

这里，Hub 就是 timer 的客户。

得到Stack如下，可以看到 hub 使用 timer 做了消息循环，于是我们需要看看 hub：

p_message
_reschedules, timer.py:127
__call__, timer.py:65
fire_timers, hub.py:142
create_loop, hub.py:300
run_once, hub.py:193
run_forever, hub.py:185
main, testUb.py:46
<module>, testUb.py:50

启动时候的逻辑如下，hub 通过 hub.timer.call_repeatedly 设置了需要调用的用户函数 fun，在 Timer 内部，fun 被包装设置为 _reschedules。

    Hub
     +
     |                                         +----------------------------------+
     |                                         |  kombu.asynchronous.timer.Timer  |
     |                                         |                                  |
     |                call_repeatedly(fun)     |                                  |
     |                                         |                                  |
     +---------------------------------------------->  _reschedules [@wraps(fun)] |
     |                                         |                                  |
     |                                         |                                  |
     |                                         |                                  |
     |                                         +----------------------------------+
     |
     |
     v

4.3.2 Hub 的使用

以下代码是Hub类，在这里，Hub 就是 timer 的用户。

可以看到，hub 建立了message_loop。在 loop 中，hub 会：

使用 fire_timers 进行 timer 处理，会设置下一次 timer。
得到 poll_timeout 后，会进行处理或者 sleep。

下面是简化版代码。

def create_loop():

    while 1:

        poll_timeout = fire_timers(propagate=propagate) if scheduled else 1

        if readers or writers:

            events = poll(poll_timeout)

            for fd, event in events or ():

                if event & READ:
                    try:
                        cb, cbargs = readers[fd]
                    try:
                        cb(*cbargs)
                    except Empty:
                        pass

        else:
            # no sockets yet, startup is probably not done.
            sleep(min(poll_timeout, 0.1))
        yield

我们再看看 fire_timers，这就是调用用户方法。

def fire_timers(self, min_delay=1, max_delay=10, max_timers=10,
                propagate=()):
    timer = self.timer
    delay = None

    if timer and timer._queue:
        for i in range(max_timers):
            delay, entry = next(self.scheduler)
            if entry is None:
                break

            entry()# 调用用户方法

    return min(delay or min_delay, max_delay)

使用Entry调用用户方法

class Entry:
    """Schedule Entry."""

    def __call__(self):
        return self.fun(*self.args, **self.kwargs)# 调用用户方法

具体逻辑如下：

+--------------------------+
|                          |
|              Hub         |
|               +          |
|               |          |                        +----------------------------------+
|               |          |                        |  kombu.asynchronous.timer.Timer  |
|               |          |                        |                                  |
|               |          |  call_repeatedly(fun)  |                                  |
|               |          |                        |                                  |
|               +---------------------------------------->  _reschedules [@wraps(fun)] |
|               |          |                        |                                  |
|               |          |                        |                                  |
|               |          |                        |                                  |
|               |          |                        +----------------------------------+
|            create_loop   |
|               +          |                                   ^
|               |          |                                   |
|               |          |                                   |
|               v          |                                   |
|                          |                                   |
|   +--->  message_loop    |                                   |
|   |           +          |                                   |
|   |           |          |                                   |
|   |           v          |        iter(self.timer)           |
|   |       fire_timers +--------------------------------------+
|   |           +          |
|   |           |          |
|   |           v          |
|   |         poll         |
|   |           +          |
|   |           |          |
|   |           v          |
|   |         sleep        |
|   |           +          |
|   |           |          |
|   +-----------+          |
+--------------------------+

在celery/utils/timer2.py中定义了Timer类实例，可以看出其继承了threading.Thread，但是居然也用kombu.asynchronous.timer。

在源码注释中有：This is only used for transports not supporting AsyncIO。

其实，就是 timer2 自己做了一个线程来做定时sleep等待，然后调用用户方法而已。

from kombu.asynchronous.timer import Entry
from kombu.asynchronous.timer import Timer as Schedule
from kombu.asynchronous.timer import logger, to_timestamp

class Timer(threading.Thread): # 扩展了 线程
    """Timer thread.

    Note:
        This is only used for transports not supporting AsyncIO.
    """

    Entry = Entry
    Schedule = Schedule

    running = False
    on_tick = None

    _timer_count = count(1)

在run方法中，会定期sleep。

def run(self):
    try:
        self.running = True
        self.scheduler = iter(self.schedule)

        while not self._is_shutdown.isSet():
            delay = self._next_entry()
            if delay:
                if self.on_tick:
                    self.on_tick(delay)
                if sleep is None:  # pragma: no cover
                    break
                sleep(delay)
        try:
            self._is_stopped.set()
        except TypeError:  # pragma: no cover
            # we lost the race at interpreter shutdown,
            # so gc collected built-in modules.
            pass
    except Exception as exc:
        sys.stderr.flush()
        os._exit(1)

在_next_entry方法中，调用用户方法，这是通过kombu.asynchronous.timer完成的。

def _next_entry(self):
    with self.not_empty:
        delay, entry = next(self.scheduler)
        if entry is None:
            if delay is None:
                self.not_empty.wait(1.0)
            return delay
    return self.schedule.apply_entry(entry)
__next__ = next = _next_entry  # for 2to3

Timer 类主要是做一些定时调度方面的工作。

Heart 组件就是使用 Timer组件进行定期调度，发送心跳 Event，告诉其他 Worker 这个 Worker 还活着。

同时，当本worker 启动，停止时候，也发送 worker-online，worker-offline 这两种消息。

6.1 Heart in Bootstep

位置在：celery/worker/consumer/heart.py。

其作用就是启动 heart 功能类。

class Heart(bootsteps.StartStopStep):
    """Bootstep sending event heartbeats.

    This service sends a ``worker-heartbeat`` message every n seconds.

    Note:
        Not to be confused with AMQP protocol level heartbeats.
    """

    requires = (Events,)

    def __init__(self, c,
                 without_heartbeat=False, heartbeat_interval=None, **kwargs):
        self.enabled = not without_heartbeat
        self.heartbeat_interval = heartbeat_interval
        c.heart = None
        super().__init__(c, **kwargs)

    def start(self, c):
        c.heart = heartbeat.Heart(
            c.timer, c.event_dispatcher, self.heartbeat_interval,
        )
        c.heart.start()

    def stop(self, c):
        c.heart = c.heart and c.heart.stop()
    shutdown = stop

6.2 Heart in Consumer

位置在：celery/worker/heartbeat.py。可以看到就是从启动之后，使用 call_repeatedly 定期发送心跳。

class Heart:
    """Timer sending heartbeats at regular intervals.

    Arguments:
        timer (kombu.asynchronous.timer.Timer): Timer to use.
        eventer (celery.events.EventDispatcher): Event dispatcher
            to use.
        interval (float): Time in seconds between sending
            heartbeats.  Default is 2 seconds.
    """

    def __init__(self, timer, eventer, interval=None):
        self.timer = timer
        self.eventer = eventer

    def _send(self, event, retry=True):
        return self.eventer.send(event, freq=self.interval, ...)

    def start(self):
        if self.eventer.enabled:
            self.tref = self.timer.call_repeatedly(
                self.interval, self._send, ('worker-heartbeat',),
            )

此时变量为：

self = {Heart} <celery.worker.heartbeat.Heart object at 0x000001D377636408>
 eventer = {EventDispatcher} <celery.events.dispatcher.EventDispatcher object at 0x000001D37765B308>
 interval = {float} 2.0
 timer = {Timer: 0} <Timer(Timer-1, stopped daemon)>
 tref = {NoneType} None
  _send_sent_signal = {NoneType} None

6.3 worker-online

当启动时候，发送 worker-online 消息。

    def start(self):
        if self.eventer.enabled:
            self._send('worker-online')
            self.tref = self.timer.call_repeatedly(
                self.interval, self._send, ('worker-heartbeat',),
            )

6.4 worker-offline

当停止时候，发送 worker-offline 消息。

    def stop(self):
        if self.tref is not None:
            self.timer.cancel(self.tref)
            self.tref = None
        if self.eventer.enabled:
            self._send('worker-offline', retry=False)

6.5 发送心跳

Heart组件会调用 eventer 来群发心跳：

eventer 是 celery.events.dispatcher.EventDispatcher;
心跳是 'worker-heartbeat' 这个 Event；

所以我们下文就要分析 celery.events.dispatcher.EventDispatcher。

    def _send(self, event, retry=True):
        if self._send_sent_signal is not None:
            self._send_sent_signal(sender=self)
        return self.eventer.send(event, freq=self.interval,
                                 active=len(active_requests),
                                 processed=all_total_count[0],
                                 loadavg=load_average(),
                                 retry=retry,
                                 **SOFTWARE_INFO)

★★★★★★关于生活和技术的思考★★★★★★

微信公众账号：罗西的思考

如果您想及时得到个人撰写文章的消息推送，或者想看看个人推荐的技术资料，敬请关注。