celery 初始化过程

consumer初始过程：

这些步骤都是celery/bootsteps.py中 StartStopStep的子类，Gossip例外， 它是ConsumerStep的子类；

他们被封装在Blueprint (蓝图？)中，

    class Blueprint(bootsteps.Blueprint):
        name = 'Consumer'
        default_steps = [
            'celery.worker.consumer:Connection',
            'celery.worker.consumer:Mingle',
            'celery.worker.consumer:Events',
            'celery.worker.consumer:Gossip',
            'celery.worker.consumer:Heart',
            'celery.worker.consumer:Control',
            'celery.worker.consumer:Tasks',
            'celery.worker.consumer:Evloop',
            'celery.worker.consumer:Agent',
        ]

在Consumer的__init__中会对Blueprint初始化：

        self.blueprint = self.Blueprint(
            app=self.app, on_close=self.on_close,
        )
        self.blueprint.apply(self, **dict(worker_options or {}, **kwargs))
  # apply调用会对blueprint里的steps这个set中的module，做类似module_import，将set内的字符串表示的module作import

  #...

  # 然后用parent, 也就是传给apply调用的 self, self也就是Consumer对象，对每个setp module生成对象实例，之后都存中blueprint对象的self.steps字典中，字典key是module字符名字

        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)

到这里完成了各个step对象实例化。

Consumer的start方法， 应该是提供给它的上层调用者， 调用开始一个Consumer的， 在start方法中，会调用blueprint的 start， blueprint的start会将封装在自己的self.steps中的step， 全部执行 step.start(parent), parent就是Consumer的对象实例：

    # Consumer的start 开始
    def start(self):
        blueprint = self.blueprint
        while blueprint.state != CLOSE:
            self.restart_count += 1
            maybe_shutdown()
            try:
                blueprint.start(self)   # 启动blueprint 开始逐个step的初始化
            except self.connection_errors as exc:
              # ....
              # 省略

    # Blueprint的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):
            self._debug('Starting %s', step.alias)
            self.started = i + 1
            step.start(parent)
            debug('^-- substep ok')          

接下来就是最上述的那些 step 对象的初始化了，他们全部都定义在 worker/consumer.py文件中， 先看Connect， 因为step.start实际就是调用每个step对象的start方法：

    # Connection的start
    # 可以看出， 之所以叫 step， 确实是名副其实， 其实就是将Consumer的初始化过程，封装成一个一个，串行的步骤，第一步是Connection
    # 这里的c， 就是start传来的parent, 那么它其实就是 Consumer对象本身
    def start(self, c):
        c.connection = c.connect()
    # 那么这里调用了Consumer的connect方法， 连接amqp，以及一些和连接相关的逻辑


    # Mingle step是进行 searching neighbors，来发现其他节点, 如果发现了node， 会和对方同步一下revoked数据

    # Event step会初始化事件调度器，代码里面经过一顿绕来绕去的， 它其实定义在 events/__init__.py里的 EventDispatcher
    # 这里是Event step的start方法：
    def start(self, c):
        # flush events sent while connection was down.
        prev = self._close(c)
        # 使用Consumer的connect返回的conn， 和本节点hostname初始化 Dispatcher对象, 这样Consumer.event_dispatcher完成初始化
        dis = c.event_dispatcher = c.app.events.Dispatcher(
            c.connect(), hostname=c.hostname,
            enabled=self.send_events, groups=self.groups,
        )
        if prev:
            dis.extend_buffer(prev)
            dis.flush()

    # Gossip step 初始化amqp Channel，通过自己的get_consumer初始化kombu.Consumer(amqp协议的consumer)
    # self.Receiver其实是 events/__init__.py内的 EventReceiver类， 那么就是初始化这个事件接收者类， routing_key这里是直接写死的
    # EventReceiver这里用做的是consumer节点之间的消息事件调度，
    # 它通过订阅routing_key是 ‘worker.#’这个topic类型的exchange type，来订阅调度消息
    # 初始化的amqp协议的consumer最终，存储在Gossip的self.consumers中
    # 在初始化kombu.Consumer的地方， 设置了mq消息的回调方法，就是 Gossip的on_message
    # 发现Gossip居然做的是election之类的事，好神奇，还没看到celery有什么election方面的功能，表示没明白是干吗的， 以后再吧
    def get_consumers(self, channel):
        self.register_timer()
        ev = self.Receiver(channel, routing_key='worker.#')
        return [kombu.Consumer(
            channel,
            queues=[ev.queue],
            on_message=partial(self.on_message, ev.event_from_message),
            no_ack=True
        )]


    # Heart step 是初始化一个heartbeat, 用于node节点之间的心跳检查
    def start(self, c):
        c.heart = heartbeat.Heart(
            c.timer, c.event_dispatcher, self.heartbeat_interval,
        )
        c.heart.start()


    # Control 用于 CELERY_ENABLE_REMOTE_CONTROL 方面的

    # Tasks step 是初始化 amqp的consumer， 是TaskConsumer, 最终这个amqp的consumer存储在 Consumer.task_consumer中
        c.task_consumer = c.app.amqp.TaskConsumer(
            c.connection, on_decode_error=c.on_decode_error,
        )

    # Evloop step 初始化事件循环器，这是consumer事件循环的关键:
    # 这里最终会运行的loop 是 self.loop = loops.asynloop if hub else loops.synloop
    # 也就是异步方式还是同步方式， 基于celery使用的方式而不同， 比如用 eventlet 或是 fork 模式
    def loop_args(self):
        return (self, self.connection, self.task_consumer,
                self.blueprint, self.hub, self.qos, self.amqheartbeat,
                self.app.clock, self.amqheartbeat_rate)    # loop调用的时候用这些args
    def start(self, c):
        self.patch_all(c)
        c.loop(*c.loop_args())

    def patch_all(self, c):
        c.qos._mutex = DummyLock()  

    # loop最终是定义在 worker/loop.py中
    # loop函数最终使用的是kombu模块的async/hub.py中的Hub对象，来完成异步的事件模型， Hup对象是在上层中初始化的
    # Consumer的self.hub，是由上层调用初始化的，其实就是上层的WorkController类， 后面来说这个
    # 它会回调具体的callback

至此Consumer的初始化完成了， 值得一提的是， Consumer的 create_task_handler 方法是创建app/task.py中Task的地方，

这个Task也就是我们在使用celery的时候， 用继承也好， 用装饰器也好，生成的具体的业务task了，

我们可以为每个task设置不同strategy策略

# 这是Consumer的 绑定task name和 strategy 函数的地方
# WorkController会调用这个update_strategies方法
# Consumer的self.strategies属性中， 会存储字典， 它是task name和 strategy的对应
    def update_strategies(self):
    loader = self.app.loader
        for name, task in items(self.app.tasks):
            self.strategies[name] = task.start_strategy(self.app, self)    # 这里最终返回的是strategy函数里的闭包函数: task_message_handler
            task.__trace__ = build_tracer(name, task, loader, self.hostname,
                                      app=self.app)

    # 在create_task_handler方法中， 会用到self.strategies， 根据amqp过来的body里的task字段获取到task name， 然后招到strategy调用
    def create_task_handler(self):
        strategies = self.strategies
        on_unknown_message = self.on_unknown_message
        on_unknown_task = self.on_unknown_task
        on_invalid_task = self.on_invalid_task
        callbacks = self.on_task_message

        # on_task_received会通过loops.py中对应的loop函数，和Hup的事件循环， 注册在Consumer.task_consumer的callbacks中
        def on_task_received(body, message):          
            try:
                name = body['task']
            except (KeyError, TypeError):
                return on_unknown_message(body, message)

            try:
                strategies[name](message, body,
                                 message.ack_log_error,
                                 message.reject_log_error,
                                 callbacks)
            except KeyError as exc:
                on_unknown_task(body, message, exc)
            except InvalidTaskError as exc:
                on_invalid_task(body, message, exc)

        return on_task_received           

这样对worker/job.py中的Request的初始化前， 可以做一些事情了, 而且可以为每个任务定义不同的策略, 策略函数task_message_handler是on_task_received触发后第一个调用的方法：

    # 这里是承接上面代码中的 strategies[name](....) 调用
    def task_message_handler(message, body, ack, reject, callbacks,
                             to_timestamp=to_timestamp):
        req = Req(body, on_ack=ack, on_reject=reject,
                  app=app, hostname=hostname,
                  eventer=eventer, task=task,
                  connection_errors=connection_errors,
                  message=message)
        # ...省略
    # 需要注意的是此处的Request的实例req， 和app/task.py的Task对象的request属性， 并不是一个东西，不要混淆    

    # 这里的Req是worker/job.py中的Request对象， 或者也可以自己利用Request对象继承后，做一些重载形成自己的Request类，在这里使用 
    # 另外， 业务的Task的在Request的 self.task = task or self.app.tasks[name]中
    # Request中的execute_using_pool方法会最终调用self.tasks， 当然看方法名就知道它并不是直接调用， 是一个池
    # execute_using_pool方法是在 WorkController 中被调用的：

    # worker/__init__.py中的WorkController的方法至于谁触发的 _process_task，需要再往上层跟才知道了
    def _process_task(self, req):
        """Process task by sending it to the pool of workers."""
        try:
            req.execute_using_pool(self.pool)
        except TaskRevokedError:
            try:
                self._quick_release()   # Issue 877
            except AttributeError:
                pass
        except Exception as exc:
            logger.critical('Internal error: %r\n%s',
                            exc, traceback.format_exc(), exc_info=True)    

WorkController 的初始化， 采用的方式是和Consumer一样的， 通过Blueprint来逐个步骤的执行

它会完成：

这些组件的初始化， 其中就是Hub和amqp的Queue，还有对上面分析的Consumer调用

default_steps = set([
    'celery.worker.components:Hub',
    'celery.worker.components:Queues',
    '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.autoreload:WorkerComponent',

])

下次再跟进去看这个。

完！