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

當(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ù)洹?/p>
剩下的只有這個吃胡蘿卜的家伙了……
當(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ù)?hellip;…恐怖。如果你懂AMQP,這些文檔還好,但是有多少人懂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”的消息需要被送到兩個隊列,“log-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)中使用。對于其他語言有一大坨材料。
但是對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。
fromamqplibimportclient_0_8asamqp conn=amqp.Connection(host="localhost:5672",userid="guest", password="guest",virtual_host="/",insist=False) chan=conn.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ī):
chan.queue_declare(queue="po_box",durable=True, exclusive=False,auto_delete=False) chan.exchange_declare(exchange="sorting_room",type="direct",durable=True, auto_delete=False,)
這段代碼干了啥?首先,它創(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 會在沒有消息的時候崩掉):
msg=chan.basic_get("po_box") printmsg.body chan.basic_ack(msg.delivery_tag)
但是如果你想要應(yīng)用程序在消息到達(dá)的時候立即得到通知怎么辦?這種情況下不能使用chan.basic_get(),你需要用chan.basic_consume()注冊一個新消息到達(dá)的回調(diào)。
defrecv_callback(msg): print\'Received:\'+msg.body chan.basic_consume(queue=\'po_box\',no_ack=True, callback=recv_callback,consumer_tag="testtag") whileTrue: chan.wait() chan.basic_cancel("testtag")
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()的實例代碼。
好了,這就是消費者的全部代碼。
不過沒有人發(fā)送消息的話,要消費者何用?所以需要一個生產(chǎn)者。下面的代碼示例表明如何將一個簡單消息發(fā)送到交換區(qū)“sorting_room”,并且標(biāo)記為路由鍵“jason” :
msg=amqp.Message("Testmessage!") msg.properties["delivery_mode"]=2 chan.basic_publish(msg,exchange="sorting_room",routing_key="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()
很簡單吧。
現(xiàn)在我們已經(jīng)寫好了生產(chǎn)者和消費者,讓他們跑起來吧。假設(shè)你的RabbitMQ在localhost上安裝并且運行。
打開一個終端,執(zhí)行python ./amqp_consumer.py讓消費者運行,并且創(chuàng)建隊列、交換機(jī)和綁定。
然后在另一個終端運行python ./amqp_publisher.py “AMQP rocks.”。如果一切良好,你應(yīng)該能夠在第一個終端看到輸出的消息。
付諸使用吧
我知道這個教程是非常粗淺的關(guān)于AMQP/RabbitMQ和如何使用Python訪問的教程。希望這個可以說明所有的概念如何在Python當(dāng)中被組合起來。同時,我很高興回答我知道的問題。【譯注:譯者也是一樣的】。接下來是,集群化(clustering)!不過我需要先把它弄懂再說。
文章名稱:[RabbitMQ+Python入門經(jīng)典]兔子和兔子窩
標(biāo)題網(wǎng)址:http://www.chinadenli.net/article48/chedhp.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供網(wǎng)站改版、電子商務(wù)、App開發(fā)、動態(tài)網(wǎng)站、網(wǎng)站策劃、網(wǎng)站建設(shè)
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請盡快告知,我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場,如需處理請聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時需注明來源: 創(chuàng)新互聯(lián)