Python操作RabbitMQ初體驗（一）

由於想用Python實現一套分散式系統，來管理和監控CDN的內容與運作狀態，誤打誤撞認識了RabbitMQ，推薦的人很多，如餘鋒《我為什麼要選擇RabbitMQ》等等。

在MQ這個詞彙映入眼簾的時候，我花了好幾個小時去閱讀什麼是MQ，也就是Message Queue(訊息佇列）。顧名思義，訊息佇列，裝滿訊息的佇列，佇列，資料結構名詞，具備先進先出特性的一種資料結構。訊息佇列無非是用來傳遞訊息的而已，那麼其意義何在，應用場景又在哪裡，具備什麼特性，及其獨特優勢是什麼，為什麼要用，這些都是浮現在我腦海中的一連串問題。

經過一大串搜索，終於膚淺的理解訊息隊列是應用對應用的訊息傳遞的溝通方式。例如需要分析多台伺服器的日誌，完全可以每台伺服器都用一個進程往一個Mysql資料庫的表裡面寫入數據，也就是所需要的信息，然後再寫幾個進程，讀取表裡數據，進行數據分析不是挺好，不過很快，這種設計的醜陋之處就顯現出來了…… 你想要多個程式從一個佇列當中取資料來處理？沒問題，我們硬編碼程式的個數好了……什麼？還要能夠允許程式動態地增加和減少的時候動態進行壓力分配？這是很流行的《RabbitMQ+Python入門經典 兔子與兔子窩》裡面的例子。想想也是，當我的CDN傳輸來一大堆資料的時候，資料的分發，處理，所有的一切都會是問題。但其實我還是沒明白，Rabbit到底是怎麼樣實現這些東西的。

概念上說，RabbitMQ是AMPQ（高階訊息協定佇列）的標準實現，據說不熟悉AMQP，會看不懂RabbitMQ的文檔。但也只能建大的理解關鍵概念了。整個RabbitMQ的實作原理模型見下圖，其實就是一個有路由任務分發佇列的生產者與消費者模型。如圖所示，即生產者生產出相應的訊息，發送給路由器，路由器根據訊息中的關鍵Key訊息，將訊息分發到不同的訊息隊列中，再由消費者去不同的訊息隊列中讀取數據的過程。

Broker：簡單來說就是訊息佇列伺服器實體。
　　Exchange：訊息交換機，它指定訊息依什麼規則，路由到哪個佇列。
　　Queue：訊息佇列載體，每個訊息都會被投入到一個或多個佇列。
　　Binding：綁定，它的作用就是把exchange和queue依照路由規則綁定起來。
　　Routing Key：路由關鍵字，exchange根據這個關鍵字進行訊息投遞。
　　vhost：虛擬主機，一個broker裡可以開立多個vhost，用來當作不同使用者的權限分離。
　　producer：訊息生產者，就是投遞訊息的程式。
　　consumer：訊息消費者，就是接受訊息的程式。
　　channel：訊息通道，在客戶端的每個連線裡，可建立多個channel，每個channel代表一個會話任務。

訊息佇列的使用過程大概如下：

（1）客戶端連接到訊息佇列伺服器，開啟一個channel。
　　（2）客戶端聲明exchange，並設定相關屬性。
　　（3）客戶端宣告一個queue，並設定相關屬性。
　　（4）客戶端使用routing key，在exchange和queue之間建立好綁定關係。
　　（5）客戶端投遞訊息到exchange。

了解了RabbitMQ大概流程與優勢之後，我開始在自己的Fedora上進行RabbitMQ的試用。

根據RabbitMQ官網介紹，安裝需要從此處下載 

rpm --import http://www.rabbitmq.com/rabbitmq-signing-key-public.asc yum install rabbitmq-server-3.3.0-1.noarch.rpm

chkconfig rabbitmq-server on

/sbin/service rabbitmq-server stop/start

grep beam.smp /var/log/audit/audit.log | audit2allow -M mypolsemodule -i mypol.pp/sbin/service rabbitmq-server start

pip install pika

#!/usr/bin/env pythonimport pikaconnection = pika.BlockingConnection(pika.ConnectionParameters(              
 'localhost'))channel = connection.channel()

channel.queue_declare(queue='sayhello')

channel.basic_publish(exchange='',                        
routing_key='hello',                                    
body='hello world')print "Sent ‘hello world'"

connection.close()

channel.queue_declare(queue='hello')

def callback(ch, method, properties, body):    print " [x] Received %r" % (body,)

channel.basic_consume(callback,                      
queue='hello',                      
no_ack=True)print ' [*] Waiting for messages. To exit press CTRL+C'channel.start_consuming()