[轉(zhuǎn)][RabbitMQ+Python入門經(jīng)典] 兔子和兔子窩

http://blog.csdn.net/linvo/article/details/5750987

?

RabbitMQ 作為一個工業(yè)級的消息隊列服務(wù)器，在其 客戶端手冊列表 的Python段當(dāng)中推薦了 一篇blog ，作為RabbitMQ+Python的入門手冊再合適不過了。不過，正如其標(biāo)題 Rabbit and Warrens （兔 子和養(yǎng)兔場）一樣，這篇英文寫的相當(dāng)俏皮，以至于對于我等非英文讀者來說不像一般的技術(shù)文檔那么好懂，所以，翻譯一下吧。翻譯過了，希望其他人可以少用一 些時間。翻譯水平有限，不可能像原文一樣俏皮，部分地方可能就意譯了，希望以容易懂為準(zhǔn)。想看看老外的幽默的，推薦去看原文，其實，也不是那么難理解……

原文： http://blogs.digitar.com/jjww/2009/01/rabbits-and-warrens/

當(dāng)時我們的動機(jī)很簡單：從生產(chǎn)環(huán)境的電子郵件處理流程當(dāng)中分支出一個特定的離線分析流程。我們開始用的MySQL，將要處理的東西放在表里面，另一個程序從中取。不過很快，這種設(shè)計的丑陋之處就顯現(xiàn)出來了…… 你想要多個程序從一個隊列當(dāng)中取數(shù)據(jù)來處理？沒問題，我們硬編碼程序的個數(shù)好了……什么？還要能夠允許程序動態(tài)地增加和減少的時候動態(tài)進(jìn)行壓力分配？

是的，當(dāng)年我們想的簡單的東西（做一個分支處理）逐漸變成了一個棘手的問題。以前拿著錘子（MySQL）看所有東西都是釘子（表）的年代是多么美好……

在搜索了一下之后，我們走進(jìn)了消息隊列（message queue）的大門。不不，我們當(dāng)然知道消息隊列是什么，我們可是以做電子郵件程序謀生的。我們實現(xiàn)過各種各樣的專業(yè)的，高速的內(nèi)存隊列用來做電子郵件處理。我們不知道的是那一大類現(xiàn)成的、通用的消息隊列（MQ）服務(wù)器——無論是用什么語言寫出的，不需要復(fù)雜的裝配的，可以自然的在網(wǎng)絡(luò)上的應(yīng)用程序之間傳送數(shù)據(jù)的一類程序。不用我們自己寫？看看再說。

讓大家看看你們的 Queue 吧 ……

過去的4年里，人們寫了有好多好多的開源的MQ服務(wù)器啊。其中大多數(shù)都是某公司例如LiveJournal寫出來用來解決特定問題的。它們的確不關(guān)心上面跑的是什么類型的消息，不過他們的設(shè)計思想通常是和創(chuàng)建者息息相關(guān)的（消息的持久化，崩潰恢復(fù)等通常不在他們考慮范圍內(nèi)）。不過，有三個專門設(shè)計用來做及其靈活的消息隊列的程序值得關(guān)注：

·???????? Apache ActiveMQ

·???????? ZeroMQ

·???????? RabbitMQ

Apache ActiveMQ 曝光率最高，不過看起來它有些問題，可能會造成丟消息。不可接受，下一個。

ZeroMQ 和 RabbitMQ 都支持一個開源的消息協(xié)議，成為AMQP。AMQP的一個優(yōu)點是它是一個靈活和開放的協(xié)議，以便和另外兩個商業(yè)化的Message Queue （IBM和Tibco）競爭，很好。不過ZeroMQ不支持消息持久化和崩潰恢復(fù)，不太好。剩下的只有RabbitMQ了。如果你不在意消息持久化和崩潰恢復(fù)，試試ZeroMQ吧，延遲很低，而且支持靈活的拓?fù)洹?

剩下的只有這個吃胡蘿卜的家伙了 ……

?

當(dāng)我讀到它是用Erlang寫的時候，RabbitMQ震了我一下。 Erlang ?是愛立信開發(fā)的高度并行的語言，用來跑在電話交換機(jī)上。是的，那些要求6個9的在線時間的東西。在Erlang當(dāng)中，充斥著大量輕量進(jìn)程，它們之間用消息傳遞來通信。聽起來思路和我們用消息隊列的思路是一樣的，不是么？

而且，RabbitMQ支持持久化。是的，如果RabbitMQ死掉了，消息并不會丟失，當(dāng)隊列重啟，一切都會回來。而且，正如在DigiTar（注：原文作者的公司）做事情期望的那樣，它 可以和Python無縫結(jié)合 。除此之外，RabbitMQ的文檔相當(dāng)?shù)摹植馈Ｈ绻愣瓵MQP，這些文檔還好，但是有多少人懂AMQP？這些文檔就像MySQL的文檔假設(shè)你已經(jīng)懂了SQL一樣……不過沒關(guān)系啦。

好了，廢話少說。這里是花了一周時間閱讀關(guān)于AMQP和關(guān)于它如何在RabbitMQ上工作的文檔之后的一個總結(jié)，還有，怎么在Python當(dāng)中使用。

開始吧

AMQP當(dāng)中有四個概念非常重要：虛擬主機(jī)（virtual host），交換機(jī)（exchange），隊列（queue）和綁定（binding）。一個虛擬主機(jī)持有一組交換機(jī)、隊列和綁定。為什么需要多個虛擬主機(jī)呢？很簡單，RabbitMQ當(dāng)中，用戶只能在虛擬主機(jī)的粒度進(jìn)行權(quán)限控制。因此，如果需要禁止A組訪問B組的交換機(jī)/隊列/綁定，必須為A和B分別創(chuàng)建一個虛擬主機(jī)。每一個RabbitMQ服務(wù)器都有一個默認(rèn)的虛擬主機(jī)“/”。如果這就夠了，那現(xiàn)在就可以開始了。

交換機(jī)，隊列，還有綁定 …… 天哪！

剛開始我思維的列車就是在這里脫軌的…… 這些鬼東西怎么結(jié)合起來的？

隊列（Queues）是你的消息（messages）的終點，可以理解成裝消息的容器。消息就一直在里面，直到有客戶端（也就是消費者，Consumer）連接到這個隊列并且將其取走為止。不過。你可以將一個隊列配置成這樣的：一旦消息進(jìn)入這個隊列，biu~，它就煙消云散了。這個有點跑題了……

需要記住的是，隊列是由消費者（Consumer）通過程序建立的，不是通過配置文件或者命令行工具。這沒什么問題，如果一個消費者試圖創(chuàng)建一個已經(jīng)存在的隊列，RabbitMQ就會起來拍拍他的腦袋，笑一笑，然后忽略這個請求。因此你可以將消息隊列的配置寫在應(yīng)用程序的代碼里面。這個概念不錯。

OK，你已經(jīng)創(chuàng)建并且連接到了你的隊列，你的消費者程序正在百無聊賴的敲著手指等待消息的到來，敲啊，敲啊…… 沒有消息。發(fā)生了什么？你當(dāng)然需要先把一個消息放進(jìn)隊列才行。不過要做這個，你需要一個交換機(jī)（Exchange）……

交換機(jī)可以理解成具有路由表的路由程序，僅此而已。每個消息都有一個稱為路由鍵（routing key）的屬性，就是一個簡單的字符串。交換機(jī)當(dāng)中有一系列的綁定（binding），即路由規(guī)則（routes），例如，指明具有路由鍵 “X” 的消息要到名為timbuku的隊列當(dāng)中去。先不討論這個，我們有點超前了。

你的消費者程序要負(fù)責(zé)創(chuàng)建你的交換機(jī) 們 （復(fù)數(shù)）。啥？你是說你可以有多個交換機(jī)？是的，這個可以有，不過為啥？很簡單，每個交換機(jī)在自己獨立的進(jìn)程當(dāng)中執(zhí)行，因此增加多個交換機(jī)就是增加多個進(jìn)程，可以充分利用服務(wù)器上的CPU核以便達(dá)到更高的效率。例如，在一個8核的服務(wù)器上，可以創(chuàng)建5個交換機(jī)來用5個核，另外3個核留下來做消息處理。類似的，在RabbitMQ的集群當(dāng)中，你可以用類似的思路來擴(kuò)展交換機(jī)一邊獲取更高的吞吐量。

OK，你已經(jīng)創(chuàng)建了一個交換機(jī)。但是他并不知道要把消息送到哪個隊列。你需要路由規(guī)則，即綁定（binding）。一個綁定就是一個類似這樣的規(guī)則：將交換機(jī)“desert（沙漠）”當(dāng)中具有路由鍵“阿里巴巴”的消息送到隊列“hideout（山洞）”里面去。換句話說，一個綁定就是一個基于路由鍵將交換機(jī)和隊列連接起來的路由規(guī)則。例如，具有路由鍵“audit”的消息需要被送到兩個隊列，“l(fā)og-forever”和“alert-the-big-dude”。要做到這個，就需要創(chuàng)建兩個綁定，每個都連接一個交換機(jī)和一個隊列，兩者都是由“audit”路由鍵觸發(fā)。在這種情況下，交換機(jī)會復(fù)制一份消息并且把它們分別發(fā)送到兩個隊列當(dāng)中。交換機(jī)不過就是一個由綁定構(gòu)成的路由表。

現(xiàn)在復(fù)雜的東西來了：交換機(jī)有多種類型。他們都是做路由的，不過接受不同類型的綁定。為什么不創(chuàng)建一種交換機(jī)來處理所有類型的路由規(guī)則呢？因為每種規(guī)則用來做匹配分子的CPU開銷是不同的。例如，一個“topic”類型的交換機(jī)試圖將消息的路由鍵與類似“ dogs.* ”的模式進(jìn)行匹配。匹配這種末端的通配符比直接將路由鍵與“ dogs ”比較（“direct”類型的交換機(jī)）要消耗更多的CPU。如果你不需要“topic”類型的交換機(jī)帶來的靈活性，你可以通過使用“direct”類型的交換機(jī)獲取更高的處理效率。那么有哪些類型，他們又是怎么處理的呢？

Fanout Exchange ?– 不處理路由鍵。你只需要簡單的將隊列綁定到交換機(jī)上。一個發(fā)送到交換機(jī)的消息都會被轉(zhuǎn)發(fā)到與該交換機(jī)綁定的所有隊列上。很像子網(wǎng)廣播，每臺子網(wǎng)內(nèi)的主機(jī)都獲得了一份復(fù)制的消息。Fanout交換機(jī)轉(zhuǎn)發(fā)消息是最快的。

Direct Exchange ?– 處理路由鍵。需要將一個隊列綁定到交換機(jī)上，要求該消息與一個特定的路由鍵完全匹配。這是一個完整的匹配。如果一個隊列綁定到該交換機(jī)上要求路由鍵 “dog”，則只有被標(biāo)記為“ dog ”的消息才被轉(zhuǎn)發(fā)，不會轉(zhuǎn)發(fā) dog.puppy ，也不會轉(zhuǎn)發(fā) dog.guard ，只會轉(zhuǎn)發(fā) dog 。

Topic Exchange ?– 將路由鍵和某模式進(jìn)行匹配。此時隊列需要綁定要一個模式上。符號“#”匹配一個或多個詞，符號“*”匹配不多不少一個詞。因此“audit.#”能夠匹配到“ audit.irs.corporate ”，但是“ audit.* ” 只會匹配到“ audit.irs ”。我在RedHat的朋友做了一張不錯的圖，來表明topic交換機(jī)是如何工作的：

?

Source:? Red Hat Messaging Tutorial: 1.3 Topic Exchange

持久化這些小東西們

你花了大量的時間來創(chuàng)建隊列、交換機(jī)和綁定，然后，砰～服務(wù)器程序掛了。你的隊列、交換機(jī)和綁定怎么樣了？還有，放在隊列里面但是尚未處理的消息們呢？

放松～如果你是用默認(rèn)參數(shù)構(gòu)造的這一切的話，那么，他們，都，biu~，灰飛煙滅了。是的，RabbitMQ重啟之后會干凈的像個新生兒。你必須重做所有的一切，亡羊補(bǔ)牢，如何避免將來再度發(fā)生此類杯具？

隊列和交換機(jī)有一個創(chuàng)建時候指定的標(biāo)志durable，直譯叫做堅固的。durable的唯一含義就是具有這個標(biāo)志的隊列和交換機(jī)會在重啟之后重新建立，它不表示說在隊列當(dāng)中的消息會在重啟后恢復(fù)。那么如何才能做到不只是隊列和交換機(jī)，還有消息都是持久的呢？

但是首先一個問題是，你真的需要消息是持久的嗎？對于一個需要在重啟之后回復(fù)的消息來說，它需要被寫入到磁盤上，而即使是最簡單的磁盤操作也是要消耗時間的。如果和消息的內(nèi)容相比，你更看重的是消息處理的速度，那么不要使用持久化的消息。不過對于我們@DigiTar來說，持久化很重要。

當(dāng)你將消息發(fā)布到交換機(jī)的時候，可以指定一個標(biāo)志“Delivery Mode”（投遞模式）。根據(jù)你使用的AMQP的庫不同，指定這個標(biāo)志的方法可能不太一樣（我們后面會討論如何用Python搞定）。簡單的說，就是將Delivery Mode設(shè)置成2，也就是持久的（persistent）即可。一般的AMQP庫都是將Delivery Mode設(shè)置成1，也就是非持久的。所以要持久化消息的步驟如下：

1.??? 將交換機(jī)設(shè)成 durable。

2.??? 將隊列設(shè)成 durable。

3.??? 將消息的 Delivery Mode 設(shè)置成2 。

就這樣，不是很復(fù)雜，起碼沒有造火箭復(fù)雜，不過也有可能犯點小錯誤。

下面還要羅嗦一個東西……綁定（Bindings）怎么辦？我們無法在創(chuàng)建綁定的時候設(shè)置成durable。沒問題，如果你綁定了一個durable的隊列和一個durable的交換機(jī)，RabbitMQ會自動保留這個綁定。類似的，如果刪除了某個隊列或交換機(jī)（無論是不是durable），依賴它的綁定都會自動刪除。

注意兩點：

·???????? RabbitMQ 不允許你綁定一個非堅固（non-durable）的交換機(jī)和一個durable的隊列。反之亦然。要想成功必須隊列和交換機(jī)都是durable的。

·???????? 一旦創(chuàng)建了隊列和交換機(jī)，就不能修改其標(biāo)志了。例如，如果創(chuàng)建了一個non-durable的隊列，然后想把它改變成durable的，唯一的辦法就是刪除這個隊列然后重現(xiàn)創(chuàng)建。因此，最好仔細(xì)檢查創(chuàng)建的標(biāo)志。

開始喂蛇了～

【譯注】說喂蛇是因為Python的圖標(biāo)是條蛇。

AMQP的一個空白地帶是如何在Python當(dāng)中使用。對于其他語言有一大坨材料。

·???????? Java –? http://www.rabbitmq.com/java-client.html

·???????? .NET –? http://www.rabbitmq.com/releases/rabbitmq-dotnet-client/v1.5.0/rabbitmq-dotnet-client-1.5.0-user-guide.pdf

·???????? Ruby –? http://somic.org/2008/06/24/ruby-amqp-rabbitmq-example/

但是對Python老兄來說，你需要花點時間來挖掘一下。所以我寫了這個，這樣別的家伙們就不需要經(jīng)歷我這種抓狂的過程了。

首先，我們需要一個Python的AMQP庫。有兩個可選：

·???????? py-amqplib ?– 通用的AMQP

·???????? txAMQP ?– 使用? Twisted ?框架的AMQP庫，因此允許異步I/O。

根據(jù)你的需求，py-amqplib或者txAMQP都是可以的。因為是基于Twisted的，txAMQP可以保證用異步IO構(gòu)建超高性能的AMQP程序。但是Twisted編程本身就是一個很大的主題……因此清晰起見，我們打算用 py-amqplib。 更新：請參見 Esteve Fernandez 關(guān)于 txAMQP 的使用和代碼樣例的 回復(fù) 。

AMQP支持在一個TCP連接上啟用多個MQ通信channel，每個channel都可以被應(yīng)用作為通信流。每個AMQP程序至少要有一個連接和一個channel。

?

?

每個channel都被分配了一個整數(shù)標(biāo)識，自動由Connection()類的.channel()方法維護(hù)。或者，你可以使用.channel(x)來指定channel標(biāo)識，其中x是你想要使用的channel標(biāo)識。通常情況下，推薦使用.channel()方法來自動分配channel標(biāo)識，以便防止沖突。

現(xiàn)在我們已經(jīng)有了一個可以用的連接和channel。現(xiàn)在，我們的代碼將分成兩個應(yīng)用，生產(chǎn)者（producer）和消費者（consumer）。我們先創(chuàng)建一個消費者程序，他會創(chuàng)建一個叫做“po_box”的隊列和一個叫“sorting_room”的交換機(jī)：

?

這段代碼干了啥？首先，它創(chuàng)建了一個名叫“ po_box ”的隊列，它是durable的（重啟之后會重新建立），并且最后一個消費者斷開的時候不會自動刪除（ auto_delete=False ）。在創(chuàng)建durable的隊列（或者交換機(jī)）的時候，將auto_delete設(shè)置成false是很重要的，否則隊列將會在最后一個消費者斷開的時候消失，與durable與否無關(guān)。如果將durable和auto_delete都設(shè)置成True，只有尚有消費者活動的隊列可以在RabbitMQ意外崩潰的時候自動恢復(fù)。

（你可以注意到了另一個標(biāo)志，稱為“exclusive”。如果設(shè)置成True，只有創(chuàng)建這個隊列的消費者程序才允許連接到該隊列。這種隊列對于這個消費者程序是私有的）。

還有另一個交換機(jī)聲明，創(chuàng)建了一個名字叫“sorting_room”的交換機(jī)。auto_delete和durable的含義和隊列是一樣的。但是，.excange_declare() 還有另外一個參數(shù)叫做type，用來指定要創(chuàng)建的交換機(jī)的類型（如前面列出的）：? fanout ,? direct ?和? topic .

到此為止，你已經(jīng)有了一個可以接收消息的隊列和一個可以發(fā)送消息的交換機(jī)。不過我們需要創(chuàng)建一個綁定，把它們連接起來。

chan.queue_bind(queue=”po_box”, exchange=”sorting_room”,
routing_key=”jason”)

這個綁定的過程非常直接。任何送到交換機(jī)“ sorting_room ”的具有路由鍵“ jason ” 的消息都被路由到名為“ po_box ” 的隊列。

現(xiàn)在，你有兩種方法從隊列當(dāng)中取出消息。第一個是調(diào)用 chan.basic_get() ，主動從隊列當(dāng)中拉出下一個消息（如果隊列當(dāng)中沒有消息，chan.basic_get()會返回None， 因此下面代碼當(dāng)中print msg.body 會在沒有消息的時候崩掉）：

?

但是如果你想要應(yīng)用程序在消息到達(dá)的時候立即得到通知怎么辦？這種情況下不能使用 chan.basic_get() ，你需要用 chan.basic_consume() 注冊一個新消息到達(dá)的回調(diào)。

?

?

chan.wait() ?放在一個無限循環(huán)里面，這個函數(shù)會等待在隊列上，直到下一個消息到達(dá)隊列。 chan.basic_cancel() ?用來注銷該回調(diào)函數(shù)。參數(shù) consumer_tag ?當(dāng)中指定的字符串和 chan.basic_consume() ?注冊的一直。在這個例子當(dāng)中 chan.basic_cancel() ?不會被調(diào)用到，因為上面是個無限循環(huán)…… 不過你需要知道這個調(diào)用，所以我把它放在了代碼里。

需要注意的另一個東西是 no_ack 參數(shù)。這個參數(shù)可以傳給 chan.basic_get() 和 chan.basic_consume() ，默認(rèn)是false。當(dāng)從隊列當(dāng)中取出一個消息的時候，RabbitMQ需要應(yīng)用顯式地回饋說已經(jīng)獲取到了該消息。如果一段時間內(nèi)不回饋，RabbitMQ會將該消息重新分配給另外一個綁定在該隊列上的消費者。另一種情況是消費者斷開連接，但是獲取到的消息沒有回饋，則RabbitMQ同樣重新分配。如果將 no_ack ?參數(shù)設(shè)置為true，則py-amqplib會為下一個AMQP請求添加一個no_ack屬性，告訴AMQP服務(wù)器不需要等待回饋。但是，大多數(shù)時候，你也許想要自己手工發(fā)送回饋，例如，需要在回饋之前將消息存入數(shù)據(jù)庫。回饋通常是通過調(diào)用 chan.basic_ack() 方法，使用消息的 delivery_tag 屬性作為參數(shù)。參見 chan.basic_get() ?的實例代碼。

好了，這就是消費者的全部代碼。（下載： amqp_consumer.py ）

不過沒有人發(fā)送消息的話，要消費者何用？所以需要一個生產(chǎn)者。下面的代碼示例表明如何將一個簡單消息發(fā)送到交換區(qū)“ sorting_room ”，并且標(biāo)記為路由鍵“ jason ” ：

?

你也許注意到我們設(shè)置消息的 delivery_mode 屬性為2，因為隊列和交換機(jī)都設(shè)置為durable的，這個設(shè)置將保證消息能夠持久化，也就是說，當(dāng)它還沒有送達(dá)消費者之前如果RabbitMQ重啟則它能夠被恢復(fù)。

剩下的最后一件事情（生產(chǎn)者和消費者都需要調(diào)用的）是關(guān)閉channel和連接：

chan.close()

conn.close()

很簡單吧。（下載： amqp_publisher.py ）

來真實地跑一下吧 ……

現(xiàn)在我們已經(jīng)寫好了生產(chǎn)者和消費者，讓他們跑起來吧。假設(shè)你的RabbitMQ在localhost上安裝并且運(yùn)行。

打開一個終端，執(zhí)行 python ./amqp_consumer.py 讓消費者運(yùn)行，并且創(chuàng)建隊列、交換機(jī)和綁定。

然后在另一個終端運(yùn)行 python ./amqp_publisher.py “AMQP rocks.” ?。如果一切良好，你應(yīng)該能夠在第一個終端看到輸出的消息。

付諸使用吧

我知道這個教程是非常粗淺的關(guān)于AMQP/RabbitMQ和如何使用Python訪問的教程。希望這個可以說明所有的概念如何在Python當(dāng)中被組合起來。如果你發(fā)現(xiàn)任何錯誤，請聯(lián)系原作者( williamsjj@digitar.com ) 【譯注：如果是翻譯問題請聯(lián)系譯者】。同時，我很高興回答我知道的問題。【譯注：譯者也是一樣的】。接下來是，集群化（clustering）！不過我需要先把它弄懂再說。

注：關(guān)于RabbitMQ的知識我主要來自這些來源，推薦閱讀：

·???????? zeromq：消息中間件分析

·???????? RabbitMQ .NET客戶端庫用戶手冊

·???????? 高級消息隊列協(xié)議（Advanced Message Queuing Protocol）：協(xié)議規(guī)約0.8 版本

–完–

?

[RabbitMQ+Python入門經(jīng)典] 兔子和兔子窩