zoukankan html css js c++ java

[源码解析] 并行分布式框架 Celery 之 worker 启动 (2)

0x00 摘要

Celery是一个简单、灵活且可靠的，处理大量消息的分布式系统，专注于实时处理的异步任务队列，同时也支持任务调度。Celery 是调用其Worker 组件来完成具体任务处理。

前文讲了 Celery 启动过程的前半部分，本文继续后半部分的分析。

本文相关其他链接如下：

[源码分析] 消息队列 Kombu 之 mailbox

[源码分析] 消息队列 Kombu 之 Hub

[源码分析] 消息队列 Kombu 之 Consumer

[源码分析] 消息队列 Kombu 之 Producer

[源码分析] 消息队列 Kombu 之启动过程

[源码解析] 消息队列 Kombu 之基本架构

以及

源码解析并行分布式框架 Celery 之架构 (2)

[源码解析] 并行分布式框架 Celery 之架构 (2)

[源码解析] 并行分布式框架 Celery 之 worker 启动 (1)

0x01 前文回顾

前文提到了，我们经过一系列过程，正式来到了 Worker 的逻辑。我们在本文将接下来继续看后续 work as a program 的启动过程。

                                     +----------------------+
      +----------+                   |  @cached_property    |
      |   User   |                   |      Worker          |
      +----+-----+            +--->  |                      |
           |                  |      |                      |
           |  worker_main     |      |  Worker application  |
           |                  |      |  celery/app/base.py  |
           v                  |      +----------------------+
 +---------+------------+     |
 |        Celery        |     |
 |                      |     |
 |  Celery application  |     |
 |  celery/app/base.py  |     |
 |                      |     |
 +---------+------------+     |
           |                  |
           |  celery.main     |
           |                  |
           v                  |
 +---------+------------+     |
 |  @click.pass_context |     |
 |       celery         |     |
 |                      |     |
 |                      |     |
 |    CeleryCommand     |     |
 | celery/bin/celery.py |     |
 |                      |     |
 +---------+------------+     |
           |                  |
           |                  |
           |                  |
           v                  |
+----------+------------+     |
|   @click.pass_context |     |
|        worker         |     |
|                       |     |
|                       |     |
|     WorkerCommand     |     |
| celery/bin/worker.py  |     |
+-----------+-----------+     |
            |                 |
            +-----------------+

为了便于大家理解，我们先给出最终的流程图如下：

0x2 Worker as a program

这里的 worker 其实就是业务主体，值得大书特书。

代码来到了celery/apps/worker.py。

class Worker(WorkController):
    """Worker as a program."""

实例化的过程调用到了WorkController基类的init。

初始化基本就是：

loader 加载各种配置；
setup_defaults做缺省设置；
setup_instance 就是正式建立，包括配置存放消息的queue。
通过Blueprint来建立 Worker 内部的各个子模块。

代码位于celery/apps/worker.py。

class WorkController:
    """Unmanaged worker instance."""

    app = None
    pidlock = None
    blueprint = None
    pool = None
    semaphore = None

    #: contains the exit code if a :exc:`SystemExit` event is handled.
    exitcode = None

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

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

    def __init__(self, app=None, hostname=None, **kwargs):
        self.app = app or self.app                      # 设置app属性
        self.hostname = default_nodename(hostname)      # 生成node的hostname
        self.startup_time = datetime.utcnow()
        self.app.loader.init_worker()                   # 调用app.loader的init_worker()方法
        self.on_before_init(**kwargs)                   # 调用该初始化方法
        self.setup_defaults(**kwargs)                   # 设置默认值
        self.on_after_init(**kwargs)
        self.setup_instance(**self.prepare_args(**kwargs))  # 建立实例

此时会调用app.loader的init_worker方法，

2.1 loader

此处的app.loader，是在Celery初始化的时候设置的loader属性，该值默认是celery.loaders.app:AppLoader。其作用就是导入各种配置。

其位于celery/loaders/base.py，定义如下：

@cached_property
def loader(self):
    """Current loader instance."""
    return get_loader_cls(self.loader_cls)(app=self)

get_loader_cls如下：

def get_loader_cls(loader):
    """Get loader class by name/alias."""
    return symbol_by_name(loader, LOADER_ALIASES, imp=import_from_cwd)

此时的loader实例就是AppLoader，然后调用该类的init_worker方法，

def init_worker(self):
    if not self.worker_initialized:             # 如果该类没有被设置过
        self.worker_initialized = True          # 设置成设置过
        self.import_default_modules()           # 导入默认的modules
        self.on_worker_init()

import_default_modules如下，主要就是导入在app配置文件中需要导入的modules，

def import_default_modules(self):
    responses = signals.import_modules.send(sender=self.app)
    # Prior to this point loggers are not yet set up properly, need to
    #   check responses manually and reraised exceptions if any, otherwise
    #   they'll be silenced, making it incredibly difficult to debug.
    for _, response in responses:   # 导入项目中需要导入的modules
        if isinstance(response, Exception):
            raise response
    return [self.import_task_module(m) for m in self.default_modules]

2.2 setup_defaults in worker

继续分析Worker类初始化过程中的self.setup_defaults方法，给运行中需要设置的参数设置值，

这之后，self.pool_cls的数值为：<class 'celery.concurrency.prefork.TaskPool'>。

代码如下：

def setup_defaults(self, concurrency=None, loglevel='WARN', logfile=None,
                   task_events=None, pool=None, consumer_cls=None,
                   timer_cls=None, timer_precision=None,
                   autoscaler_cls=None,
                   pool_putlocks=None,
                   pool_restarts=None,
                   optimization=None, O=None,  # O maps to -O=fair
                   statedb=None,
                   time_limit=None,
                   soft_time_limit=None,
                   scheduler=None,
                   pool_cls=None,              # XXX use pool
                   state_db=None,              # XXX use statedb
                   task_time_limit=None,       # XXX use time_limit
                   task_soft_time_limit=None,  # XXX use soft_time_limit
                   scheduler_cls=None,         # XXX use scheduler
                   schedule_filename=None,
                   max_tasks_per_child=None,
                   prefetch_multiplier=None, disable_rate_limits=None,
                   worker_lost_wait=None,
                   max_memory_per_child=None, **_kw):
    either = self.app.either                # 从配置文件中获取，如果获取不到就使用给定的默认值 
    self.loglevel = loglevel                # 设置日志等级
    self.logfile = logfile                  # 设置日志文件

    self.concurrency = either('worker_concurrency', concurrency)        # 设置worker的工作进程数
    self.task_events = either('worker_send_task_events', task_events)   # 设置时间
    self.pool_cls = either('worker_pool', pool, pool_cls)               # 连接池设置
    self.consumer_cls = either('worker_consumer', consumer_cls)         # 消费类设置
    self.timer_cls = either('worker_timer', timer_cls)                  # 时间类设置
    self.timer_precision = either(
        'worker_timer_precision', timer_precision,
    )
    self.optimization = optimization or O                               # 优先级设置
    self.autoscaler_cls = either('worker_autoscaler', autoscaler_cls) 
    self.pool_putlocks = either('worker_pool_putlocks', pool_putlocks)
    self.pool_restarts = either('worker_pool_restarts', pool_restarts)
    self.statedb = either('worker_state_db', statedb, state_db)         # 执行结果存储
    self.schedule_filename = either(
        'beat_schedule_filename', schedule_filename,
    )                                                                   # 定时任务调度设置
    self.scheduler = either('beat_scheduler', scheduler, scheduler_cls) # 获取调度类
    self.time_limit = either(
        'task_time_limit', time_limit, task_time_limit)                 # 获取限制时间值
    self.soft_time_limit = either(
        'task_soft_time_limit', soft_time_limit, task_soft_time_limit,
    )
    self.max_tasks_per_child = either(
        'worker_max_tasks_per_child', max_tasks_per_child,
    )                                                                   # 设置每个子进程最大处理任务的个数
    self.max_memory_per_child = either(
        'worker_max_memory_per_child', max_memory_per_child,
   )                                                                   # 设置每个子进程最大内存值
    self.prefetch_multiplier = int(either(
        'worker_prefetch_multiplier', prefetch_multiplier,
    ))
    self.disable_rate_limits = either(
        'worker_disable_rate_limits', disable_rate_limits,
    )
    self.worker_lost_wait = either('worker_lost_wait', worker_lost_wait)

2.3 setup_instance in worker

执行完成后，会继续执行 self.setup_instance方法来建立实例。

def setup_instance(self, queues=None, ready_callback=None, pidfile=None,
                   include=None, use_eventloop=None, exclude_queues=None,
                   **kwargs):
    self.pidfile = pidfile                              # pidfile
    self.setup_queues(queues, exclude_queues)           # 指定相关的消费与不消费队列
    self.setup_includes(str_to_list(include))           # 获取所有的task任务

    # Set default concurrency
    if not self.concurrency:                            # 如果没有设置默认值
        try:
            self.concurrency = cpu_count()              # 设置进程数与cpu的个数相同
        except NotImplementedError: 
            self.concurrency = 2                        # 如果获取失败则默认为2

    # Options
    self.loglevel = mlevel(self.loglevel)               # 设置日志等级
    self.ready_callback = ready_callback or self.on_consumer_ready  # 设置回调函数

    # this connection won't establish, only used for params
    self._conninfo = self.app.connection_for_read()     
    self.use_eventloop = (
        self.should_use_eventloop() if use_eventloop is None
        else use_eventloop          
    )                                                   # 获取eventloop类型
    self.options = kwargs

    signals.worker_init.send(sender=self)               # 发送信号

    # Initialize bootsteps
    self.pool_cls = _concurrency.get_implementation(self.pool_cls)  # 获取缓冲池类
    self.steps = []                                     # 需要执行的步骤
    self.on_init_blueprint() 
    self.blueprint = self.Blueprint(
        steps=self.app.steps['worker'],
        on_start=self.on_start,
        on_close=self.on_close,
        on_stopped=self.on_stopped,
    )                                                  # 初始化blueprint
    self.blueprint.apply(self, **kwargs)               # 调用初始化完成的blueprint类的apply方法

其中setup_queues和setup_includes所做的工作如下，

def setup_queues(self, include, exclude=None):
    include = str_to_list(include)
    exclude = str_to_list(exclude)
    try:
        self.app.amqp.queues.select(include)        # 添加队列消费
    except KeyError as exc:
        raise ImproperlyConfigured(
            SELECT_UNKNOWN_QUEUE.strip().format(include, exc))
    try:
        self.app.amqp.queues.deselect(exclude)      # 不消费指定的队列中的任务
    except KeyError as exc:
        raise ImproperlyConfigured(
            DESELECT_UNKNOWN_QUEUE.strip().format(exclude, exc))
    if self.app.conf.worker_direct:
        self.app.amqp.queues.select_add(worker_direct(self.hostname))  # 添加消费的队列

def setup_includes(self, includes):
    # Update celery_include to have all known task modules, so that we
    # ensure all task modules are imported in case an execv happens.
    prev = tuple(self.app.conf.include)                             # 获取配置文件中的task
    if includes:
        prev += tuple(includes)
        [self.app.loader.import_task_module(m) for m in includes]   # 将task添加到loader的task中
    self.include = includes                     
    task_modules = {task.__class__.__module__
                    for task in values(self.app.tasks)}             # 获取已经注册的任务
    self.app.conf.include = tuple(set(prev) | task_modules)         # 去重后重新设置include

2.3.1 setup_queues

self.app.amqp.queues.select(include)会设置queues。

堆栈如下：

__init__, amqp.py:59
Queues, amqp.py:259
queues, amqp.py:572
__get__, objects.py:43
setup_queues, worker.py:172
setup_instance, worker.py:106
__init__, worker.py:99
worker, worker.py:326
caller, base.py:132
new_func, decorators.py:21
invoke, core.py:610
invoke, core.py:1066
invoke, core.py:1259
main, core.py:782
start, base.py:358
worker_main, base.py:374

代码位于：celery/app/amqp.py

class Queues(dict):
    """Queue name⇒ declaration mapping.
    """

    #: If set, this is a subset of queues to consume from.
    #: The rest of the queues are then used for routing only.
    _consume_from = None

    def __init__(self, queues=None, default_exchange=None,
                 create_missing=True, autoexchange=None,
                 max_priority=None, default_routing_key=None):
        dict.__init__(self)
        self.aliases = WeakValueDictionary()
        self.default_exchange = default_exchange
        self.default_routing_key = default_routing_key
        self.create_missing = create_missing
        self.autoexchange = Exchange if autoexchange is None else autoexchange
        self.max_priority = max_priority
        if queues is not None and not isinstance(queues, Mapping):
            queues = {q.name: q for q in queues}
        queues = queues or {}
        for name, q in queues.items():
            self.add(q) if isinstance(q, Queue) else self.add_compat(name, **q)

所以目前如下：

                                     +----------------------+
      +----------+                   |  @cached_property    |
      |   User   |                   |      Worker          |
      +----+-----+            +--->  |                      |
           |                  |      |                      |
           |  worker_main     |      |  Worker application  |
           |                  |      |  celery/app/base.py  |
           v                  |      +----------+-----------+
 +---------+------------+     |                 |
 |        Celery        |     |                 |
 |                      |     |                 |
 |  Celery application  |     |                 v
 |  celery/app/base.py  |     |  +--------------+--------------+    +---> app.loader.init_worker
 |                      |     |  | class Worker(WorkController)|    |
 +---------+------------+     |  |                             |    |
           |                  |  |                             +--------> setup_defaults
           |  celery.main     |  |    Worker as a program      |    |
           |                  |  |   celery/apps/worker.py     |    |
           v                  |  +-----------------------------+    +---> setup_instance +-----> setup_queues  +------>  app.amqp.queues
 +---------+------------+     |
 |  @click.pass_context |     |
 |       celery         |     |
 |                      |     |
 |                      |     |
 |    CeleryCommand     |     |
 | celery/bin/celery.py |     |
 |                      |     |
 +---------+------------+     |
           |                  |
           |                  |
           |                  |
           v                  |
+----------+------------+     |
|   @click.pass_context |     |
|        worker         |     |
|                       |     |
|                       |     |
|     WorkerCommand     |     |
| celery/bin/worker.py  |     |
+-----------+-----------+     |
            |                 |
            +-----------------+

手机如图：

2.4 Blueprint

下面就要建立 Worker 内部的各个子模块。

worker初始化过程中，其内部各个子模块的执行顺序是由一个BluePrint类定义，并且根据各个模块之间的依赖进行排序（实际上把这种依赖关系组织成了一个 DAG）执行，此时执行的操作就是加载blueprint类中的default_steps。

具体逻辑是：

self.claim_steps方法功能是获取定义的steps
_finalize_steps 获取step依赖，并进行拓扑排序，返回按依赖关系排序的steps。
根据依赖，返回依赖排序后step。
当step生成之后，就开始调用来生成组件。
apply函数最后，返回worker。当所有的类初始化完成后，此时就是一个worker就初始化完成。

代码如下：celery/apps/worker.py。此时的Blueprint类定义在Worker里面。

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

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

celery Worker 中各个模块定为step，具体如下：

Timer：用于执行定时任务的 Timer，和 Consumer 那里的 timer 不同；
Hub：Eventloop 的封装对象；
Pool：构造各种执行池（线程/进程/协程）；
Autoscaler：用于自动增长或者 pool 中工作单元；
StateDB：持久化 worker 重启区间的数据（只是重启）；
Autoreloader：用于自动加载修改过的代码；
Beat：创建 Beat 进程，不过是以子进程的形式运行（不同于命令行中以 beat 参数运行）；

celery 中定义依赖关系主要依靠了几个类属性 requires，label，conditon，last，比如Hub依赖Timer，Consumer最后执行。

class Hub(bootsteps.StartStopStep):
    """Worker starts the event loop."""
    requires = (Timer,)

class Consumer(bootsteps.StartStopStep):
    """Bootstep starting the Consumer blueprint."""
    last = True

2.5 Blueprint基类

apply调用的是基类代码。其基类位于celery/bootsteps.py。

class Blueprint:
    """Blueprint containing bootsteps that can be applied to objects.
    """

    GraphFormatter = StepFormatter
    name = None
    state = None
    started = 0
    default_steps = set()
    state_to_name = {
        0: 'initializing',
        RUN: 'running',
        CLOSE: 'closing',
        TERMINATE: 'terminating',
    }

    def __init__(self, steps=None, name=None,
                 on_start=None, on_close=None, on_stopped=None):
        self.name = name or self.name or qualname(type(self))
        self.types = set(steps or []) | set(self.default_steps)
        self.on_start = on_start
        self.on_close = on_close
        self.on_stopped = on_stopped
        self.shutdown_complete = Event()
        self.steps = {}

apply代码如下，这个是在WorkController初始化时执行的。

def apply(self, parent, **kwargs):
    """Apply the steps in this blueprint to an object.

    This will apply the ``__init__`` and ``include`` methods
    of each step, with the object as argument::

        step = Step(obj)
        ...
        step.include(obj)

    For :class:`StartStopStep` the services created
    will also be added to the objects ``steps`` attribute.
    """
    self._debug('Preparing bootsteps.')
    order = self.order = []                         # 用于存放依序排列的需要执行的模块类
    steps = self.steps = self.claim_steps()         # 获得定义的step，生成{step.name, step}结构

    self._debug('Building graph...')
    for S in self._finalize_steps(steps):                   # 进行依赖排序后，返回对应的step
        step = S(parent, **kwargs)                          # 获得实例化的step
        steps[step.name] = step                             # 已step.name为key,step实例为val
        order.append(step)                                  # 添加到order列表中
    self._debug('New boot order: {%s}',
                ', '.join(s.alias for s in self.order))
    for step in order:                             # 遍历order列表
        step.include(parent)                       # 各个step依次执行include函数，执行create函数
    return self

2.5.1 获取定义的steps

其中self.claim_steps方法功能是获取定义的steps，具体如下：

def claim_steps(self):
    return dict(self.load_step(step) for step in self.types)# 导入types中的类，并返回已名称和类对应的k:v字典

def load_step(self, step):
    step = symbol_by_name(step)
    return step.name, step

其中self.types可以在初始化的时候传入。

def __init__(self, steps=None, name=None,
             on_start=None, on_close=None, on_stopped=None):
    self.name = name or self.name or qualname(type(self))
    self.types = set(steps or []) | set(self.default_steps)
    self.on_start = on_start
    self.on_close = on_close
    self.on_stopped = on_stopped
    self.shutdown_complete = Event()
    self.steps = {}

在Blueprint初始化时没有传入steps，所以此时types就是default_steps属性，该值就是WorkController类中的Blueprint类的default_steps值。

default_steps = {
    'celery.worker.components:Hub',
    'celery.worker.components:Pool',
    'celery.worker.components:Beat',
    'celery.worker.components:Timer',
    'celery.worker.components:StateDB',
    'celery.worker.components:Consumer',
    'celery.worker.autoscale:WorkerComponent',
}

2.5.2 _finalize_steps

_finalize_steps作用为：获取step依赖，并进行拓扑排序，返回按依赖关系排序的steps。

其中重点分析一下self._finalize_steps(steps)函数，排序操作主要在此函数中完成。

def _finalize_steps(self, steps):
        last = self._find_last()                                # 找到last=True的step，上文也提到Consumer类
        self._firstpass(steps)                                  # 将step及所需要的依赖加入到steps列表，并获取依赖
        it = ((C, C.requires) for C in values(steps))           # 数据结构((a,[b,c]),(b,[e]))
        G = self.graph = DependencyGraph(                       # 依赖图模型初始化，添加定点和边界
            it, formatter=self.GraphFormatter(root=last),
        )
        if last:                                                # last最终执行模块，依赖所有模块执行完成后执行，所以为所有模块添加last到step边界
            for obj in G:
                if obj != last:
                    G.add_edge(last, obj)
        try:
            return G.topsort()                                  # 进行拓扑运算，获得排好序的steps列表
        except KeyError as exc:
            raise KeyError('unknown bootstep: %s' % exc)

首先定义了一个拓扑类DependencyGraph，根据依赖关系添加定点和边结构：

顶点就是各个step。
边结构就是依赖step生成的列表。
结构为{step1:[step2,step3]}。

下面我们一起看一下DependencyGraph类中的初始化操作

@python_2_unicode_compatible
class DependencyGraph(object):

    def __init__(self, it=None, formatter=None):
        self.formatter = formatter or GraphFormatter()
        self.adjacent = {}                                              # 存储图形结构
        if it is not None:
            self.update(it)

    def update(self, it):
        tups = list(it)
        for obj, _ in tups:
            self.add_arc(obj)
        for obj, deps in tups:
            for dep in deps:
                self.add_edge(obj, dep)

    def add_arc(self, obj):                                               # 添加定点
        self.adjacent.setdefault(obj, [])

    def add_edge(self, A, B):                                             # 添加边界
        self[A].append(B)

图形结构已经生成了各个模块之间的依赖关系图，主要依靠celery自己实现的一套DAG，依靠拓扑排序方法，得到执行顺序。拓扑排序就是实现依赖排序，生成模块的执行顺序，然后按照循序执行模块。

2.5.3 返回依赖排序后step

这部分作用就是根据依赖，返回依赖排序后step。

此时由于这几个类中存在相互依赖的执行，比如Hub类，

class Hub(bootsteps.StartStopStep):
    """Worker starts the event loop."""

    requires = (Timer,)

Hub类就依赖于Timer类，所以_finalize_steps的工作就是将被依赖的类先导入。

此时继续分析到order列表，该列表就是所有依赖顺序解决完成后的各个类的列表，并且这些steps类都是直接继承或间接继承自bootsteps.Step。

@with_metaclass(StepType)
class Step(object):
    ...

该类使用了元类，继续查看StepType。

class StepType(type):
    """Meta-class for steps."""

    name = None
    requires = None

    def __new__(cls, name, bases, attrs):
        module = attrs.get('__module__')                                # 获取__module__属性
        qname = '{0}.{1}'.format(module, name) if module else name      # 如果获取到了__module__就设置qname为模块.name的形式，否则就设置成name
        attrs.update(
            __qualname__=qname,
            name=attrs.get('name') or qname,
        )                                                               # 将要新建的类中的属性,name和__qualname__更新为所给的类型
        return super(StepType, cls).__new__(cls, name, bases, attrs)

    def __str__(self):
        return bytes_if_py2(self.name)

    def __repr__(self):
        return bytes_if_py2('step:{0.name}{{{0.requires!r}}}'.format(self))

这里使用了有关Python元类编程的相关知识，通过在新建该类实例的时候控制相关属性的值，从而达到控制类的相关属性的目的。此时会调用Step的include方法。

def _should_include(self, parent):
    if self.include_if(parent):
        return True, self.create(parent)
    return False, None

def include(self, parent):
    return self._should_include(parent)[0]

如果继承的是StartStopStep，则调用的include方法如下，

def include(self, parent):
    inc, ret = self._should_include(parent)
    if inc:
        self.obj = ret
        parent.steps.append(self)
    return inc

排序之后，变量数据举例如下：

order = {list: 7} 
 0 = {Timer} <step: Timer>
 1 = {Hub}  
 2 = {Pool} <step: Pool>
 3 = {WorkerComponent} <step: Autoscaler>
 4 = {StateDB} <step: StateDB>
 5 = {Beat} <step: Beat>
 6 = {Consumer} <step: Consumer>
 __len__ = {int} 7
 
 
steps = {dict: 7} 
 'celery.worker.components.Timer' = {Timer} <step: Timer>
 'celery.worker.components.Pool' = {Pool} <step: Pool>
 'celery.worker.components.Consumer' = {Consumer} <step: Consumer>
 'celery.worker.autoscale.WorkerComponent' = {WorkerComponent} <step: Autoscaler>
 'celery.worker.components.StateDB' = {StateDB} <step: StateDB>
 'celery.worker.components.Hub' = {Hub} <step: Hub>
 'celery.worker.components.Beat' = {Beat} <step: Beat>
 __len__ = {int} 7

2.5.4 生成组件

当step生成之后，就开始调用来生成组件。

for step in order:
    step.include(parent)

对于每个组件，会继续调用。各个step依次执行include函数，执行create函数

def include(self, parent):
    return self._should_include(parent)[0]

如果需要调用，就建立组件，比如self为timer，parent是worker实例

<class 'celery.worker.worker.WorkController'>。

代码如下：

def _should_include(self, parent):
    if self.include_if(parent):
        return True, self.create(parent)
    return False, None
    
def include_if(self, w):
    return w.use_eventloop

以timer为例，

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

    def create(self, w):
        if w.use_eventloop:
            # does not use dedicated timer thread.
            w.timer = _Timer(max_interval=10.0)

其堆栈如下：

create, components.py:36
_should_include, bootsteps.py:335
include, bootsteps.py:339
apply, bootsteps.py:211
setup_instance, worker.py:139
__init__, worker.py:99
worker, worker.py:326
caller, base.py:132
new_func, decorators.py:21
invoke, core.py:610
invoke, core.py:1066
invoke, core.py:1259
main, core.py:782
start, base.py:358
worker_main, base.py:374
<module>, myTest.py:26

2.5.5 返回worker

apply函数最后，返回worker。当所有的类初始化完成后，此时就是一个worker就初始化完成。

def apply(self, parent, **kwargs):
    """Apply the steps in this blueprint to an object.

    This will apply the ``__init__`` and ``include`` methods
    of each step, with the object as argument::

        step = Step(obj)
        ...
        step.include(obj)

    For :class:`StartStopStep` the services created
    will also be added to the objects ``steps`` attribute.
    """
    self._debug('Preparing bootsteps.')
    order = self.order = []
    steps = self.steps = self.claim_steps()

    self._debug('Building graph...')
    for S in self._finalize_steps(steps):
        step = S(parent, **kwargs)
        steps[step.name] = step
        order.append(step)
    self._debug('New boot order: {%s}',
                ', '.join(s.alias for s in self.order))
    for step in order:
        step.include(parent)
    return self

self如下：

self = {Blueprint} <celery.worker.worker.WorkController.Blueprint object at 0x7fcd3ad33d30>
 GraphFormatter = {type} <class 'celery.bootsteps.StepFormatter'>
 alias = {str} 'Worker'
 default_steps = {set: 7} {'celery.worker.components:Hub', 'celery.worker.components:Consumer', 'celery.worker.components:Beat', 'celery.worker.autoscale:WorkerComponent', 'celery.worker.components:StateDB', 'celery.worker.components:Timer', 'celery.worker.components:Pool'}
 graph = {DependencyGraph: 7} celery.worker.autoscale.WorkerComponent(3)
     celery.worker.components.Pool(2)
          celery.worker.components.Hub(1)
               celery.worker.components.Timer(0)
celery.worker.components.StateDB(0)
celery.worker.components.Hub(1)
     celery.worker.components.Timer(0)
celery.worker.components.Consumer(12)
     celery.worker.autoscale.WorkerComponent(3)
          celery.worker.components.Pool(2)
               celery.worker.components.Hub(1)
                    celery.worker.components.Timer(0)
     celery.worker.components.StateDB(0)
     celery.worker.components.Hub(1)
          celery.worker.components.Timer(0)
     celery.worker.components.Beat(0)
     celery.worker.components.Timer(0)
     celery.worker.components.Pool(2)
          celery.worker.components.Hub(1)
               celery.worker.components.Timer(0)
celery.worker.components.Beat(0)
celery.worker.components.Timer(0)
celery.worker.components.Pool(2)
     celery.worker.components.Hub(1)
          celery.worker.co...
 name = {str} 'Worker'
 order = {list: 7} [<step: Timer>, <step: Hub>, <step: Pool>, <step: Autoscaler>, <step: StateDB>, <step: Beat>, <step: Consumer>]
 shutdown_complete = {Event} <threading.Event object at 0x7fcd3ad33b38>
 started = {int} 0
 state = {NoneType} None
 state_to_name = {dict: 4} {0: 'initializing', 1: 'running', 2: 'closing', 3: 'terminating'}
 steps = {dict: 7} {'celery.worker.autoscale.WorkerComponent': <step: Autoscaler>, 'celery.worker.components.StateDB': <step: StateDB>, 'celery.worker.components.Hub': <step: Hub>, 'celery.worker.components.Consumer': <step: Consumer>, 'celery.worker.components.Beat': <step: Beat>, 'celery.worker.components.Timer': <step: Timer>, 'celery.worker.components.Pool': <step: Pool>}
 types = {set: 7} {'celery.worker.autoscale:WorkerComponent', 'celery.worker.components:StateDB', 'celery.worker.components:Hub', 'celery.worker.components:Consumer', 'celery.worker.components:Beat', 'celery.worker.components:Timer', 'celery.worker.components:Pool'}

此时如下：

                                     +----------------------+
      +----------+                   |  @cached_property    |
      |   User   |                   |      Worker          |
      +----+-----+            +--->  |                      |
           |                  |      |                      |
           |  worker_main     |      |  Worker application  |
           |                  |      |  celery/app/base.py  |
           v                  |      +----------+-----------+
 +---------+------------+     |                 |
 |        Celery        |     |                 |
 |                      |     |                 |
 |  Celery application  |     |                 v
 |  celery/app/base.py  |     |  +--------------+--------------+    +---> app.loader.init_worker
 |                      |     |  | class Worker(WorkController)|    |
 +---------+------------+     |  |                             |    |
           |                  |  |                             +--------> setup_defaults
           |  celery.main     |  |    Worker as a program      |    |
           |                  |  |   celery/apps/worker.py     |    |
           v                  |  +-----------------------------+    +---> setup_instance +-----> setup_queues  +------>  app.amqp.queues
 +---------+------------+     |                                                +
 |  @click.pass_context |     |                                                |
 |       celery         |     |                 +------------------------------+
 |                      |     |                 |       apply
 |                      |     |                 |
 |    CeleryCommand     |     |                 v
 | celery/bin/celery.py |     |
 |                      |     |  +-------------------------------------+        +---------------------+     +--->  claim_steps
 +---------+------------+     |  | class Blueprint(bootsteps.Blueprint)|        |  class Blueprint    |     |
           |                  |  |                                     +------>-+                     | +------->  _finalize_steps
           |                  |  |                                     |        |                     |     |
           |                  |  |     celery/apps/worker.py           |        | celery/bootsteps.py |     |                   +--> Timer
           v                  |  +-------------------------------------+        +---------------------+     +--->  include +--->+
+----------+------------+     |                                                                                                 +--> Hub
|   @click.pass_context |     |                                                                                                 |
|        worker         |     |                                                                                                 +--> Pool
|                       |     |                                                                                                 |
|                       |     |                                                                                                 +--> ......
|     WorkerCommand     |     |                                                                                                 |
| celery/bin/worker.py  |     |                                                                                                 +--> Consumer
+-----------+-----------+     |
            | 1  app.Worker   |
            +-----------------+

手机如下：

0x3 start in worker 命令

此时初始化完成后就执行到了celery/bin/worker.py中，开始执行 worker。

worker.start()

具体回忆之前代码下：

def worker(ctx, hostname=None, pool_cls=None, app=None, uid=None, gid=None,
           loglevel=None, logfile=None, pidfile=None, statedb=None,
           **kwargs):
    """Start worker instance.
    """
    app = ctx.obj.app

    if kwargs.get('detach', False):
        return detach(...)

    worker = app.Worker(...)
    
    worker.start()         #我们回到这里
    return worker.exitcode

3.1 start in Worker as a program

此时调用的方法就是：celery/worker/worker.py。可以看到，直接就是调用 blueprint的 start函数，就是启动 blueprint 之中各个组件。

def start(self):
    try:
        self.blueprint.start(self)           # 此时调用blueprint的start方法
    except WorkerTerminate:
        self.terminate()
    except Exception as exc:
        self.stop(exitcode=EX_FAILURE)
    except SystemExit as exc:
        self.stop(exitcode=exc.code)
    except KeyboardInterrupt:
        self.stop(exitcode=EX_FAILURE)

3.2 start in blueprint

代码是：celery/bootsteps.py

此时，parent.steps就是在step.include中添加到该数组中，parent.steps目前值为[Hub,Pool,Consumer]，此时调用了worker的on_start方法。本例本例如下:

parent.steps = {list: 3} 
 0 = {Hub} <step: Hub>
 1 = {Pool} <step: Pool>
 2 = {Consumer} <step: Consumer>

具体 start 代码如下：

class Blueprint:
    """Blueprint containing bootsteps that can be applied to objects.
    """

    def start(self, parent):
        self.state = RUN
        if self.on_start:
            self.on_start()
        for i, step in enumerate(s for s in parent.steps if s is not None):
            self.started = i + 1
            step.start(parent)

3.2.1 回调 on_start in Worker

blueprint首先回调 on_start in Worker。

此时代码是/celery/apps/worker.py

具体是：

设置app；
用父类的on_start；
打印启动信息；
注册相应的信号处理handler；
做相关配置比如重定向；

class Worker(WorkController):
    """Worker as a program."""

    def on_start(self):
        app = self.app                                                  # 设置app
        WorkController.on_start(self)                                   # 调用父类的on_start

        # this signal can be used to, for example, change queues after
        # the -Q option has been applied.
        signals.celeryd_after_setup.send(
            sender=self.hostname, instance=self, conf=app.conf,
        )

        if self.purge:
            self.purge_messages()

        if not self.quiet:
            self.emit_banner()                                     # 打印启动信息

        self.set_process_status('-active-')
        self.install_platform_tweaks(self)                         # 注册相应的信号处理handler
        if not self._custom_logging and self.redirect_stdouts:
            app.log.redirect_stdouts(self.redirect_stdouts_level)

        # TODO: Remove the following code in Celery 6.0
        # This qualifies as a hack for issue #6366.
        warn_deprecated = True
        config_source = app._config_source
        if isinstance(config_source, str):
            # Don't raise the warning when the settings originate from
            # django.conf:settings
            warn_deprecated = config_source.lower() not in [
                'django.conf:settings',
            ]

3.2.2 基类on_start

此时代码是/celery/apps/worker.py。

def on_start(self):
    app = self.app
    WorkController.on_start(self)

WorkController代码在：celery/worker/worker.py。

父类的on_start就是创建pid文件。

def on_start(self):
    if self.pidfile:
        self.pidlock = create_pidlock(self.pidfile)

3.2.3 信号处理handler

其中注册相关的信号处理handler的函数如下，

def install_platform_tweaks(self, worker):
    """Install platform specific tweaks and workarounds."""
    if self.app.IS_macOS:
        self.macOS_proxy_detection_workaround()

    # Install signal handler so SIGHUP restarts the worker.
    if not self._isatty:
        # only install HUP handler if detached from terminal,
        # so closing the terminal window doesn't restart the worker
        # into the background.
        if self.app.IS_macOS:
            # macOS can't exec from a process using threads.
            # See https://github.com/celery/celery/issues#issue/152
            install_HUP_not_supported_handler(worker)
        else:
            install_worker_restart_handler(worker)                      # 注册重启的信号 SIGHUP
    install_worker_term_handler(worker)             
    install_worker_term_hard_handler(worker)
    install_worker_int_handler(worker)                                  
    install_cry_handler()                                               # SIGUSR1 信号处理函数
    install_rdb_handler()                                               # SIGUSR2 信号处理函数

单独分析一下重启的函数，

def _reload_current_worker():
    platforms.close_open_fds([
        sys.__stdin__, sys.__stdout__, sys.__stderr__,
    ])# 关闭已经打开的文件描述符
    os.execv(sys.executable, [sys.executable] + sys.argv)# 重新加载该程序

以及：

def restart_worker_sig_handler(*args):
    """Signal handler restarting the current python program."""
    set_in_sighandler(True)
    safe_say('Restarting celery worker ({0})'.format(' '.join(sys.argv)))
    import atexit
    atexit.register(_reload_current_worker)                 # 注册程序退出时执行的函数
    from celery.worker import state
    state.should_stop = EX_OK                               # 设置状态
platforms.signals[sig] = restart_worker_sig_handler

其中的platforms.signals类设置了setitem方法，

def __setitem__(self, name, handler):
    """Install signal handler.

    Does nothing if the current platform has no support for signals,
    or the specified signal in particular.
    """
    try:
        _signal.signal(self.signum(name), handler)
    except (AttributeError, ValueError):
        pass

此时就将相应的handler设置到了运行程序中，_signal就是导入的signal库。

3.2.4 逐次调用step

然后，blueprint逐次调用step的start。

此时，parent.steps就是在step.include中添加到该数组中，parent.steps目前值为[Hub,Pool,Consumer]，此时调用了worker的on_start方法，

parent.steps = {list: 3} 
 0 = {Hub} <step: Hub>
 1 = {Pool} <step: Pool>
 2 = {Consumer} <step: Consumer>

此时就继续执行Blueprint的start方法，

依次遍历parent.steps的方法中，依次遍历Hub,Pool,Consumer，调用step的start方法。

def start(self, parent):
    self.state = RUN                                        # 设置当前运行状态                          
    if self.on_start:                                       # 如果初始化是传入了该方法就执行该方法
        self.on_start()
    for i, step in enumerate(s for s in parent.steps if s is not None):     # 依次遍历step并调用step的start方法
        self._debug('Starting %s', step.alias)
        self.started = i + 1
        step.start(parent)
        logger.debug('^-- substep ok')

3.2.4.1 Hub

由于Hub重写了start方法，该方法什么都不执行，

def start(self, w):
    pass

3.2.4.2 Pool

继续调用Pool方法，此时会调用到StartStopStep,此时的obj就是调用create方法返回的对象，此时obj为pool实例,

def start(self, parent):
    if self.obj:
        return self.obj.start()

具体我们后文讲解。

3.2.4.3 Consumer

继续调用Consumer的start方法，

def start(self):
    blueprint = self.blueprint
    while blueprint.state != CLOSE:                           # 判断当前状态是否是关闭
        maybe_shutdown()                                      # 通过标志判断是否应该关闭
        if self.restart_count:                                # 如果设置了重启次数
            try:
                self._restart_state.step()                    # 重置
            except RestartFreqExceeded as exc:
                crit('Frequent restarts detected: %r', exc, exc_info=1)
                sleep(1)
        self.restart_count += 1                               # 次数加1
        try: 
            blueprint.start(self)                             # 调用开始方法
        except self.connection_errors as exc:
            # If we're not retrying connections, no need to catch
            # connection errors
            if not self.app.conf.broker_connection_retry:
                raise
            if isinstance(exc, OSError) and exc.errno == errno.EMFILE:
                raise  # Too many open files
            maybe_shutdown()
            if blueprint.state != CLOSE:                              # 如果状态不是关闭状态
                if self.connection:
                    self.on_connection_error_after_connected(exc)
                else:
                    self.on_connection_error_before_connected(exc)
                self.on_close()
                blueprint.restart(self)                               # 调用重启方法

Consumer也有blueprint，具体step如下：

Connection：管理和 broker 的 Connection 连接
Events：用于发送监控事件
Agent：cell actor
Mingle：不同 worker 之间同步状态用的
Tasks：启动消息 Consumer
Gossip：消费来自其他 worker 的事件
Heart：发送心跳事件（consumer 的心跳）
Control：远程命令管理服务

此时又进入到了blueprint的start方法，该blueprint的steps值是由Consumer在初始化的时候传入的。

传入的steps是Agent,Connection,Evloop,Control,Events,Gossip,Heart,Mingle,Tasks类的实例，然后根据调用最后添加到parent.steps中的实例就是[Connection,Events,Heart,Mingle,Tasks,Control,Gossip,Evloop]，此时就依次调用这些实例的start方法。

首先分析一下Connection的start方法，

def start(self, c):
    c.connection = c.connect()
    info('Connected to %s', c.connection.as_uri())

就是调用了consumer的connect()函数，

def connect(self):
    """Establish the broker connection.

    Retries establishing the connection if the
    :setting:`broker_connection_retry` setting is enabled
    """
    conn = self.app.connection_for_read(heartbeat=self.amqheartbeat)        # 心跳

    # Callback called for each retry while the connection
    # can't be established.
    def _error_handler(exc, interval, next_step=CONNECTION_RETRY_STEP):
        if getattr(conn, 'alt', None) and interval == 0:
            next_step = CONNECTION_FAILOVER
        error(CONNECTION_ERROR, conn.as_uri(), exc,
              next_step.format(when=humanize_seconds(interval, 'in', ' ')))

    # remember that the connection is lazy, it won't establish
    # until needed.
    if not self.app.conf.broker_connection_retry:                           # 如果retry禁止
        # retry disabled, just call connect directly.
        conn.connect()                                                      # 直接连接
        return conn                                                         # 返回conn

    conn = conn.ensure_connection(
        _error_handler, self.app.conf.broker_connection_max_retries,
        callback=maybe_shutdown,
    )                                                                       # 确保连接上
    if self.hub:
        conn.transport.register_with_event_loop(conn.connection, self.hub)  # 使用异步调用
    return conn                                                             # 返回conn

此时就建立了连接。

继续分析Task的start方法，

def start(self, c):
    """Start task consumer."""
    c.update_strategies()                                           # 更新已知的任务

    # - RabbitMQ 3.3 completely redefines how basic_qos works..
    # This will detect if the new qos smenatics is in effect,
    # and if so make sure the 'apply_global' flag is set on qos updates.
    qos_global = not c.connection.qos_semantics_matches_spec

    # set initial prefetch count
    c.connection.default_channel.basic_qos(
        0, c.initial_prefetch_count, qos_global,
    )                                                               # 设置计数

    c.task_consumer = c.app.amqp.TaskConsumer(
        c.connection, on_decode_error=c.on_decode_error,
    )                                                               # 开始消费

    def set_prefetch_count(prefetch_count):
        return c.task_consumer.qos(
            prefetch_count=prefetch_count,
            apply_global=qos_global,
        )
    c.qos = QoS(set_prefetch_count, c.initial_prefetch_count)       # 设置计数

此时就开启了对应的任务消费，开启消费后我们继续分析一下loop的开启，

def start(self, c):
    self.patch_all(c)
    c.loop(*c.loop_args())

此时就是调用了consumer中的loop函数，该loop函数就是位于celery/worker/loops.py中的asyncloop函数。

stack如下：

asynloop, loops.py:43
start, consumer.py:592
start, bootsteps.py:116
start, consumer.py:311
start, bootsteps.py:365
start, bootsteps.py:116
start, worker.py:203
worker, worker.py:327
caller, base.py:132
new_func, decorators.py:21
invoke, core.py:610
invoke, core.py:1066
invoke, core.py:1259
main, core.py:782
start, base.py:358
worker_main, base.py:374

asyncloop函数如下：

def asynloop(obj, connection, consumer, blueprint, hub, qos,
         heartbeat, clock, hbrate=2.0):
    """Non-blocking event loop."""                                  # 其中obj就是consumer实例
    RUN = bootsteps.RUN                                             # 获取运行状态
    update_qos = qos.update
    errors = connection.connection_errors

    on_task_received = obj.create_task_handler()                    # 创建任务处理头

    _enable_amqheartbeats(hub.timer, connection, rate=hbrate)       # 定时发送心跳包

    consumer.on_message = on_task_received                          # 设置消费的on_message为on_task_received
    consumer.consume()                                              # 开始消费
    obj.on_ready()                                                  # 调用回调函数
    obj.controller.register_with_event_loop(hub)                    # 向所有生成的blueprint中的实例注册hub
    obj.register_with_event_loop(hub)                               

    # did_start_ok will verify that pool processes were able to start,
    # but this will only work the first time we start, as
    # maxtasksperchild will mess up metrics.
    if not obj.restart_count and not obj.pool.did_start_ok():
        raise WorkerLostError('Could not start worker processes')

    # consumer.consume() may have prefetched up to our
    # limit - drain an event so we're in a clean state
    # prior to starting our event loop.
    if connection.transport.driver_type == 'amqp':
        hub.call_soon(_quick_drain, connection)

    # FIXME: Use loop.run_forever
    # Tried and works, but no time to test properly before release.
    hub.propagate_errors = errors
    loop = hub.create_loop()                                        # 创建loop,本质是一个生成器

    try:
        while blueprint.state == RUN and obj.connection:            # 检查是否在运行并且连接是否有
            # shutdown if signal handlers told us to.
            should_stop, should_terminate = (
                state.should_stop, state.should_terminate,
            )
            # False == EX_OK, so must use is not False
            if should_stop is not None and should_stop is not False:
                raise WorkerShutdown(should_stop)
            elif should_terminate is not None and should_stop is not False:
                raise WorkerTerminate(should_terminate)

            # We only update QoS when there's no more messages to read.
            # This groups together qos calls, and makes sure that remote
            # control commands will be prioritized over task messages.
            if qos.prev != qos.value:
                update_qos()                                        

            try:
                next(loop)                                          # 循环下一个
            except StopIteration:
                loop = hub.create_loop()
    finally:
        try:
            hub.reset()
        except Exception as exc:  # pylint: disable=broad-except
            logger.exception(
                'Error cleaning up after event loop: %r', exc)

至此，异步Loop就开启了，然后就开始了服务端的事件等待处理，worker流程就启动完毕。

0x4 本文总结

本文主要讲述了worker的启动流程。celery默认启动的就是以进程的方式启动任务，然后异步IO处理消费端的事件，至此worker的启动流程概述已分析完毕。

                                     +----------------------+
      +----------+                   |  @cached_property    |
      |   User   |                   |      Worker          |
      +----+-----+            +--->  |                      |
           |                  |      |                      |
           |  worker_main     |      |  Worker application  |
           |                  |      |  celery/app/base.py  |
           v                  |      +----------+-----------+
 +---------+------------+     |                 |
 |        Celery        |     |                 |
 |                      |     |                 |
 |  Celery application  |     |                 v
 |  celery/app/base.py  |     |  +--------------+--------------+    +---> app.loader.init_worker
 |                      |     |  | class Worker(WorkController)|    |
 +---------+------------+     |  |                             |    |
           |                  |  |                             +--------> setup_defaults
           |  celery.main     |  |    Worker as a program      |    |
           |                  |  |   celery/apps/worker.py     |    |
           v                  |  +-----------------------------+    +---> setup_instance +-----> setup_queues  +------>  app.amqp.queues
 +---------+------------+     |                                                +
 |  @click.pass_context |     |                                                |
 |       celery         |     |                 +------------------------------+
 |                      |     |                 |       apply
 |                      |     |                 |
 |    CeleryCommand     |     |                 v
 | celery/bin/celery.py |     |
 |                      |     |  +-------------------------------------+        +---------------------+     +--->  claim_steps
 +---------+------------+     |  | class Blueprint(bootsteps.Blueprint)| apply  |  class Blueprint    |     |
           |                  |  |                                     +------>-+                     | +------->  _finalize_steps
           |                  |  |                                     |        |                     |     |
           |                  |  |     celery/apps/worker.py           |        | celery/bootsteps.py |     |                   +--> Timer
           v                  |  +-----+---------------+---------------+        +---------------------+     +--->  include +--->+
+----------+------------+     |        ^               |                                                                        +--> Hub
|   @click.pass_context |     |        |               | start                                                                  |
|        worker         |     |        |               |                                                                        +--> Pool
|                       |     |        |               |        +---->  WorkController.on_start                                 |
|                       |     |        |               |        |                                                               +--> ......
|     WorkerCommand     |     |        |               +------------->   Hub. start / Pool.start /  Task.start                  |
| celery/bin/worker.py  |     |        |                        |                                                               +--> Consumer
+-----------+-----------+     |        |                        +---->   Evloop.start +----->  asynloop
            | 1  app.Worker   |        |
            +-----------------+        |
            |                          |
            | 2 blueprint.start        |
            +------------------------->+

手机如下：

0xFF 参考

Celery 源码学习（二）多进程模型

celery原理初探

celery源码分析-wroker初始化分析(上)

celery源码分析-worker初始化分析（下）https://blog.csdn.net/qq_33339479/article/details/80958059

celery worker初始化--DAG实现

python celery多worker、多队列、定时任务

celery 详细教程-- Worker篇

使用Celery

Celery 源码解析一：Worker 启动流程概述

Celery 源码解析二：Worker 的执行引擎

Celery 源码解析三：Task 对象的实现

Celery 源码解析四：定时任务的实现

Celery 源码解析五：远程控制管理

Celery 源码解析六：Events 的实现

Celery 源码解析七：Worker 之间的交互

Celery 源码解析八：State 和 Result

查看全文

相关阅读:
Passing Reference by value
WPF中override ResourceDictionary中的设置的方法
 WPF中TextBox的PreviewMouseLeftButtonUp事件
 Attribute的理解和认识
 IIS添加服务
 Unix时间戳转换成C#中的DateTime
KMP算法的实现
 IDA 宏定义
 实验吧-catalyst-system
python整数转ASCII码

原文地址：https://www.cnblogs.com/rossiXYZ/p/14563777.html