今天要聊的話題可能被大家關(guān)注得不過(guò)，但是對(duì)于 Celery 來(lái)說(shuō)確實(shí)很有用的功能，曾經(jīng)我在工作中遇到這類情況，就是我們將所有的任務(wù)都放在同一個(gè)隊(duì)列里面，然后有一天突然某個(gè)同學(xué)的代碼寫得不對(duì)，導(dǎo)致大量的耗時(shí)任務(wù)被同時(shí)塞進(jìn)了消息隊(duì)列里面，這就悲劇了，這直接導(dǎo)致了其他服務(wù)長(zhǎng)時(shí)間不可用，例如發(fā)送登錄短信驗(yàn)證碼無(wú)法使用了，還有支付信息無(wú)法同步了等等，反正就是造成了一些不小的影響。

當(dāng)時(shí)我們的處理方式就很被動(dòng)，只能手動(dòng)連接上 MQ，然后把消息卸掉，其實(shí)也就手動(dòng)將這些消息拋棄掉，從而讓其他業(yè)務(wù)的消息可能正常運(yùn)行。但是，這種方式也只適合當(dāng)初作為少量流量的情況，對(duì)于搭建了大集群和大量任務(wù)的消息隊(duì)列來(lái)說(shuō)，這種方式是不可想象的，這么做是要死人的，不僅僅是被累垮，上頭的口水都能把你淹了。所以，這個(gè)時(shí)候，我需要介紹一個(gè) Celery 不太常被人使用的功能——遠(yuǎn)程控制。

遠(yuǎn)程控制功能

其實(shí) Celery 很早之前就存在控制命令，例如可以使用 Python shell 的 shell 命令，可以查看任務(wù)狀態(tài)的 status 命令等等，但是這些命令都是本地的，不能讓人覺(jué)得有意思，但是，這里有兩個(gè)系列的命令很厲害，它們分別是：

inspect：主要用于查看 Celery 狀態(tài)信息

control：主要用于設(shè)置 Celery 狀態(tài)

例如，我在機(jī)器 A 運(yùn)行著一個(gè) Celery，機(jī)器 B 也運(yùn)行著一個(gè) Celery，機(jī)器 C 沒(méi)有運(yùn)行 Celery，但是我可以在機(jī)器 C 上查詢機(jī)器 A 或者機(jī)器 B 上的任務(wù)狀態(tài)，甚至可以刪除和停止任務(wù)，這些都是很簡(jiǎn)單可以實(shí)現(xiàn)的，但是本文不是講解這些功能的文章，而是解析這些功能的文章，所以有興趣的同學(xué)可以參考這份官方文檔繼續(xù)了解。

遠(yuǎn)程控制功能組件

要實(shí)現(xiàn)遠(yuǎn)程控制功能，我們需要從宏觀上先看看 Celery 的設(shè)計(jì)思路，在 Celery 中，采用的是分布式的管理方式，其實(shí)沒(méi)有太大秘密，就是每個(gè)節(jié)點(diǎn)之間都是通過(guò)廣播/單播進(jìn)行通信，從而達(dá)到協(xié)同效果，但是這過(guò)程還是有很多不好之處的，值得我們來(lái)思考一番。

Celery 每個(gè)運(yùn)行的實(shí)例都維護(hù)著一個(gè) Control Node，其實(shí)就是一個(gè)可以接收/發(fā)送消息的對(duì)象，這個(gè)對(duì)象的封裝是 Kombu.pidbox.Mailbox.Node，我們就先來(lái)看看創(chuàng)建的實(shí)現(xiàn)吧。

還是回到第一篇，在 Consumer 的 Blueprint 中，有一個(gè)叫做 Control 的 Bootstep，這個(gè)就是用于節(jié)點(diǎn)管理和通信的，我們來(lái)看一下：

?

其實(shí)代碼還是比較簡(jiǎn)單的，有兩個(gè)地方值得我們關(guān)注，分別是：

Line 25：這里是構(gòu)建了一個(gè) Pidbox

Line 26 - Line 28：這個(gè) Bootstep 的 start、stop 和 shutdown 方法都是使用的 box 的

所以這個(gè) Pidbox 是什么就很重要了，在我們看之前，不放看下上面的注釋，也許會(huì)更容易一些：

?

雖然這里關(guān)系說(shuō)得很明確了，但是我們還是有必要看看的，畢竟有可能里面有設(shè)置什么特殊的東西：

?

ok，確實(shí)還好，很誠(chéng)實(shí)得就是用 Kombu 的 mailbox，但是 kombu 的 mailbox 是什么，可能很多人都沒(méi)試過(guò)，我之前也沒(méi)試過(guò)，后來(lái)試了一下，感覺(jué)還挺有意思的，注意，下面這段可能是現(xiàn)在互聯(lián)網(wǎng)上公開(kāi)的為數(shù)不多的可以運(yùn)行得 Kombu Demo 示例，甚至于講解。

Kombu Mailbox

在 Kombu 中提供了 Mailbox 的實(shí)現(xiàn)，它的作用就是通過(guò) Mailbox 我們可以實(shí)現(xiàn)不同實(shí)例之間的消息發(fā)送和處理，具體可以是單播 和 廣播，這個(gè)在 Celery 中是作為 Control 的功能使用的，但是，在其他的模型里面，例如 Celery 試圖實(shí)現(xiàn)但是沒(méi)有實(shí)現(xiàn)的 Actor 模型里面也是可以用的。

Anyway，下面還是講講 Kombu 中的 Mailbox 是怎么用的吧，當(dāng)初找相關(guān)的資料費(fèi)了老大力了，但是，并沒(méi)有太大收獲，所以自己總結(jié)了一番。在 Kombu 中，Mailbox 中只有一個(gè)概念，那就是 Node

Node：每個(gè) Node 都是一個(gè)實(shí)例，互相直接沒(méi)有關(guān)聯(lián)，可以完全獨(dú)立，他們通過(guò) mailbox 進(jìn)行通信

但是，為了測(cè)試，我們還會(huì)引入一個(gè) client 的概念，但是這個(gè)概念不是 Kombu 自己的，而是我為了演示效果添加進(jìn)來(lái)的，所以現(xiàn)在我們應(yīng)該有兩個(gè)地方，分別是：Node 和 Client，其中可以認(rèn)為 Node 是 Server 端，Client 是觸發(fā)端，你會(huì)發(fā)現(xiàn)，Client 只是做了一件觸發(fā)的工作，沒(méi)有其他更多的事情：

?

這是 node 的代碼，你會(huì)發(fā)現(xiàn)它底層其實(shí)還是依賴于 Kombu 的 Connection，所以可以看到依賴的還是我們 Celery 里面的 Broker，這點(diǎn)很重要。然后再看看我們是怎么觸發(fā)的：

?

可以發(fā)現(xiàn)這里非常簡(jiǎn)單，還是通過(guò) Connection 構(gòu)建出 mailbox，有一點(diǎn)需要注意的就是，Broker 要一致，不然你讓他們?cè)趺赐ㄐ?？?zhí)行這段代碼，然后你就會(huì)在 node 上看到執(zhí)行效果了，具體怎樣，體驗(yàn)之后就明了。

Celery 的遠(yuǎn)程控制

看過(guò) Kombu 的實(shí)現(xiàn)例子之后，我們來(lái)看看 Celery 是怎么構(gòu)建這些對(duì)象的，首先還是得從最開(kāi)始的 control 開(kāi)始說(shuō)起，control 在 Celery 中也是有兩處的，一處是 app/control.py，另外一處就是：woker/control.py，可以認(rèn)為第一處的是對(duì)外的接口，而第二處的是初始化的入口，實(shí)現(xiàn)自然就是 Kombu 提供的了，這里只是用到他們而已。

所以，現(xiàn)在來(lái)看，我們的目標(biāo)很簡(jiǎn)單了，無(wú)非是看

如何初始化 mailbox 和 Node 的

提供了那些對(duì)外接口可以使用的

下面就這兩個(gè)問(wèn)題進(jìn)行一一解答

mailbox 的初始化

故事是從 control 這個(gè) Bootstep 開(kāi)始說(shuō)起，這是是初始化的起源：

?

這段代碼我們前面已經(jīng)見(jiàn)過(guò)了，同時(shí)我們也已經(jīng)知道了 self.box 是個(gè)啥了，但是，對(duì)于更進(jìn)一步的 c.app.control.mailbox.Node 中的 c.app.control.mailbox 是啥還不知道，不妨來(lái)看看：

?

ok，這里也很清晰，因?yàn)檫@個(gè) control_cls 就是我們后面要看的：

control_cls = ‘celery.app.control:Control’

所以 Mailbox 也就是 Kombu 的 Mailbox 了，這里沒(méi)做什么改動(dòng)。除此之外，還有一個(gè)地方需要我們?nèi)リP(guān)注的，那就是收到消息后怎么處理，這個(gè)得看到 Bootstep 的 start 操作，這里是初始化過(guò)程中會(huì)被調(diào)用到的：

?

start 的第一句（Line 50）沒(méi)毛病，因?yàn)橛玫亩际?Celery 的 Connection，然后是第二句，這里我們根據(jù)之前的例子，已經(jīng)很清楚會(huì)發(fā)生什么事情了，所以關(guān)鍵就是 Line 37 中的 on_message，每當(dāng)有其他消息過(guò)來(lái)的時(shí)候，這里都是處理點(diǎn)。

從 Line 42 來(lái)看，Celery 還是甩鍋給了 Kombu，但是這也不是啥問(wèn)題，所以我們得找 Kombu 問(wèn)清楚它是怎么處理的：

?

ok，這里一直有個(gè)叫 clock 的東西，我先不看，后面再說(shuō)一下，先看看 dispatch 是如何處理的：

?

在 Line 99 這里其實(shí)是通過(guò)是否有發(fā)送主體（參數(shù)傳過(guò)來(lái)的 reply_to）來(lái)判斷是 單播 還是 多播 的消息，然后選擇不同的處理方式，分別是 Line 118 中的 handle_call 用來(lái)處理 單播，而 Line 121 中的 handler_cast 用來(lái)處理 多播。這里有一個(gè)點(diǎn)，那就是 Line 116 中的 handlers 里面放置了所有注冊(cè)的函數(shù)的信息，這個(gè)我們稍后會(huì)看到。

接口的注冊(cè)

前面說(shuō)了，Celery 注冊(cè)了很多管理接口給我們使用，我們就看看有那些注冊(cè)接口以及這些接口是如何注冊(cè)進(jìn)去的，我們是否可以自定義管理接口。關(guān)于接口注冊(cè)相關(guān)的代碼，我們得走到 celery/worker/control.py 中，現(xiàn)在進(jìn)來(lái)看一看：

?

這里有兩個(gè)注冊(cè)函數(shù)，其實(shí)也就是我們前面說(shuō)過(guò)的對(duì)應(yīng)的兩類操作，分別是 查看 和 設(shè)置 類，然后也可以發(fā)現(xiàn)，其實(shí)注冊(cè)就是往 Panel 這個(gè) Dict 里頭寫入一些 key 和 value 對(duì)，然而，這里有兩個(gè) Dict 是需要我們關(guān)注的，他們分別是：

data：key 就是名稱，value 是處理函數(shù)

meta：key 是名稱，value 是元數(shù)據(jù)，整個(gè)數(shù)據(jù)描述為：

?

ok，了解完這些我們就知道了遠(yuǎn)程命令的對(duì)象和處理函數(shù)的對(duì)應(yīng)關(guān)系都放在 data 和 meta 里面，這有什么用？回想一下之前 Mailbox 的構(gòu)造函數(shù)的地方：

?

注意看 Line 28，用的就是這里的 data，然后就直接用來(lái)構(gòu)造 Node 了，現(xiàn)在和前面的關(guān)系對(duì)應(yīng)起來(lái)，了解了吧？

下面我就找個(gè)復(fù)雜點(diǎn)的例子看看，是怎么講一個(gè)函數(shù)注冊(cè)進(jìn) data 這個(gè) Dict 里面的：

?

這個(gè)功能很明確了，在注釋中已經(jīng)提及了，但是，我們并不 care 它的功能，我們更多的是關(guān)注它是怎么發(fā)生的，在 Line 227 這里就是調(diào)用注冊(cè)函數(shù)進(jìn)行注冊(cè)，可以看到，args 分別對(duì)應(yīng)到我們下面的幾個(gè)命名參數(shù)，然后調(diào)用 control_command 之后其實(shí)就是直接掛在 Dict 上了，沒(méi)有其他操作，需要注意的是 Line 51 的 if args 這個(gè)條件，我在整個(gè) Celery 中都沒(méi)有看到有使用，所以應(yīng)該這里是預(yù)留的。

重要：有一點(diǎn)值得注意的是，前面也有稍微提到，Celery 的分布式實(shí)現(xiàn)機(jī)制是廣播，所以我們?cè)趩螜C(jī)上發(fā)送的命令，只要沒(méi)有指定主機(jī)，那么都是以廣播的形式發(fā)送出去，所有的實(shí)例都將受到這個(gè)消息，然后根據(jù)消息處理本機(jī)的事務(wù)，所以我們?cè)诳创a的時(shí)候需要著重關(guān)注這一理念。

遠(yuǎn)程控制客戶端

關(guān)于控制消息接收和處理的邏輯我們已經(jīng)看完了，那么我們來(lái)看看我們?cè)诿钚兄星孟旅畹臅r(shí)候，這一切是怎么運(yùn)行起來(lái)的。要看這些邏輯，我推薦的入口是：celery/bin/control.py，這是一條典型的 Celery 命令類，這里的結(jié)構(gòu)就比較復(fù)雜了，我不多說(shuō)，直接看最后的結(jié)果，那就是調(diào)用的時(shí)候：

編輯：hfy