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

序列文章：

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

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

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

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

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

• Celery 源码解析六：Events 的实现

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

• Celery 源码解析八：State 和 Result

前面对于 Celery 的分布式处理已经做了一些介绍，例如第五章的 远程控制 和第六章的 Event机制，但是，我认为这些分布式都比较简单，并没有体现出多实例之间的协同作用，所以，今天就来点更加复杂的，对于多实例直接的交互更多，这就是 Gossip 和 Mingle。

Mingle

在 Celery 的介绍中，Mingle 主要用在启动或者重启的时候，它会和其他的 worker 交互，从而进行同步。同步的数据有：

• 其他 worker 的 clock
• 其他 worker 已经处理掉的 tasks

这其实也就是它的所有功能的，所以你可以猜测功能应该很简单吧？不妨一起来看看，最开始还是回忆一下第一篇文章中的 Bootstep，所以我们可以毫无压力得找出源码所在的文件：



这里从注释中可以很简单得看出 Mingle 的作用，然后初始化也是比较简单，关键还是 Line 37 的 start，需要我们关注 sync 做了什么，为什么上来二话不说就 sync？其实上来就 sync 很好理解，毕竟 Mingle 的作用就是进行 sync 嘛，所以我们要关注的是如何实现的：



这里原来的代码有点冗余，我给忽略掉了，直接上精简后的代码，所以你可以很清晰得看到代码的逻辑是这样的：

1. Mingle 向每一个 Worker 发送问候：hello
2. 每个 Worker 都向 Mingle 回复自己的信息（clock 和 tasks）
3. Mingle 更新自己的信息

这些逻辑我们从精简后的代码可以简单看出来，所以就不细说了，但是有一点需要展开讲讲，那就是 Line 47 中的 inspect.hello，这应该是 第五篇 的内容，但是，之前只是介绍了一下如何注册，并没有对这些命令一一解析，所以这里用到了，我们就不妨看看里面的内容。



ok，这里我们可以看到在 Line 319、320 就返回了两个东西，分别是：

• revoked：当前 worker 记录的已被完成的 tasks
• clock：当前 worker 的 clock

然后就返回到刚起来的 worker 了，收到这个消息的 worker 就根据这两个信息刷新自己的状态，然后继续运行，Mingle 也就完成了自己的任务了。

Gossip

和 Mingle 不同，Gossip 却是消费 Event 的，本来按道理应该放在 第六篇 中介绍，但是由于篇幅原因，所以一起放在这里来说了，不多赘述，我们直接看 Bootstep：



由于 Gossip 的初始化内容太多，所以我也不全都展开了，挑了些重点（还是很多），但是目前我们可以忽略大部分的内容，最先需要关注的是 Line 24，如果你够细心的话你会发现这个 Bootstep 和其他不一样，因为它继承的是 ConsumerStep，这是会注册一个 Consumer 的！

然后我们没啥好看了，所以按照套路还是看看 start 呗，然而它调用的也是父类的 start，所以，没办法咯，直接跟过去：



诺，是这样吧，是增加了 Consumer，这样的话，我们就必须看看这个 Consumer 是什么了，能够消费什么样的数据：



好，这样就清晰了，所有关于 worker.# 的 Event 都被这里消费了，这里算是看完了。

那现在的问题变成了这些 Event 都是从哪里过来的，我们有必要对源头进行一下追踪，但是，怎么追踪呢？回想一下 第六篇 中讲 Event 的消息传递的那里，再和这里一对比，事情就很清楚了。

你以为 Gossip 就这么结束了么？嘿嘿，那你就被 Celery 给蒙骗了，悄悄告诉你，Celery 在 Gossip 中埋伏了一个厉害的功能，但是没有对外宣称，那就是 Leader 选举！，不信？我带你去看看：



这是选举的入口，先不解析代码，我们先来看看有谁调用了它：



ok，可以发现这有个 control 命令用到了它，这里有注释，我们可以看到参数分别代表的意思：

• id：唯一的标识，用于识别一次选举
• topic：本次选举的 topic，其实是标识 action 的类型
• action：本次选举的目的，选中的 leader 负责处理这个 action

那么这样我们就清楚了，首先，有一个 action 需要执行，但是，那么多的 worker，交给谁执行呢？这就需要进行 选举，那么选举的方式是怎么进行的呢，我先用一张图来描述一下这个过程：

1. control 表示需要进行一个选举，然后一个 worker 的 gossip 就发送了一个 Event：worker-elect，然后所有的 Worker 都能接收到：

   

2. 每个 Worker 接收到之后，就对这个选举进行响应，将自己的选号（clock）送过去，这样，每个 Worker 在发送选号的同时，也接收到别人的选号，因为收发的路线太多，我就找一个 Worker 来表示收，但是其他 Worker 也是有收的，只是我没有标出来：

   

3. 当一个 Worker 收到所有 Worker 的 ACK 之后，那么它就会对所有的 Worker 的 选号 进行排序，选出其中 最大的选号 作为本次选举的 Leader，如果 Leader 是自己那么就处理这个 Action，如果不是自己，那么忽略，应该被选中的 Leader 也在执行这个过程，所以不需要别人担心。

   

这就是实际执行的示意图，对应到代码就分别是：

1. 第一步中的 Control 要求选举和发送选举 Event 我们前面已经看过了

2. Worker 收到选举 Event 之后，发出自己的参选声明：

   

3. 每个 Wroker 对别人回应的参选信息进行选举：

   

ok，整个流程就是这样的了，那么问题来了，万一有一个 Worker 收不到 replies 或者发出的 reply 不小心丢了会怎么样？是不是整个选举过程就进行不下去了？我好像没有看到 Celery 有在这方面做一些努力。