RabbitMQ 學習
一、概要
一個不錯的入門教程： http://blog.csdn.net/linvo/article/details/5750987 寫的挺好的，只是剛開始看可能不太懂，模模糊糊，多看幾遍，試著寫點代碼之后，再看。就比較清晰了。
官方文檔使用了 using the pika 0.9.8 Python client 。本文使用 http://github.com/celery/py-amqp amqp 1.4.6
至于安裝，自己找下教程吧。不難，先安裝 Erlang，再安裝RabbitMQ。然后配置一下，有個web控制臺。之后就是python編程使用了。
再加一個不錯的中文資料： http://blog.chinaunix.net/topic/surpershi/
MQ全稱為Message Queue, 消息隊列（MQ）是一種應用程序對應用程序的通信方法。RabbitMQ是流行的開源消息隊列系統，用erlang語言開發。RabbitMQ是AMQP（高級消息隊列協議）的標準實現。
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。
exchange接收到消息后，就根據消息的key和已經設置的binding，進行消息路由，將消息投遞到一個或多個隊列里。
exchange也有幾個類型，完全根據key進行投遞的叫做Direct交換機，例如，綁定時設置了 routing key為”abc”，那么客戶端提交的消息，只有設置了key為”abc”的才會投遞到隊列。對key進行模式匹配后進行投遞的叫做Topic交換機，符 號”#”匹配一個或多個詞，符號”*”匹配正好一個詞。例如”abc.#”匹配”abc.def.ghi”，”abc.*”只匹配”abc.def”。還 有一種不需要key的，叫做Fanout交換機，它采取廣播模式，一個消息進來時，投遞到與該交換機綁定的所有隊列。
RabbitMQ支持消息的持久化，也就是數據寫在磁盤上，為了數據安全考慮，我想大多數用戶都會選擇持久化。消息隊列持久化包括3個部分：
（1）exchange持久化，在聲明時指定durable => 1
（2）queue持久化，在聲明時指定durable => 1
（3）消息持久化，在投遞時指定delivery_mode => 2（1是非持久化）
如果exchange和queue都是持久化的，那么它們之間的binding也是持久化的。如果exchange和queue兩者之間有一個持久化，一個非持久化，就不允許建立綁定。
二、基本使用
入門教程看會之后，就差不多了。
下面示例代碼：
consumer 消費者
# amqp_consumer.py
# -*- coding: utf-8 -*-
__author__ = 'lpe234'
__date__ = '2014-12-15'
import amqp
conn = amqp.Connection(host="localhost:5672", userid="guest", password="guest", virtual_host="/", insist=False)
chan = conn.channel()
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, )
chan.queue_bind(queue="po_box", exchange="sorting_room", routing_key="1111")
def receive_callback(msg):
print 'Received: ' + msg.body + ' from channel #' + str(msg.channel.channel_id)
chan.basic_consume(queue='po_box', no_ack=True, callback=receive_callback, consumer_tag="consumer")
while True:
chan.wait()
chan.basic_cancel("consumer")
chan.close()
conn.close()
producer 生產者
# amqp_publisher.py
# -*- coding: utf-8 -*-
__author__ = 'lpe234'
__date__ = '2014-12-15'
import amqp
import json
conn = amqp.Connection(host="localhost:5672", userid="guest", password="guest", virtual_host="/", insist=False)
chan = conn.channel()
for x in xrange(10):
msg = json.dumps({'id': str(x)+'111', 'lists': [{'id': 12345}, {'id': 12345}, {'id': 15656}, {'id': '4545'}, ]})
print msg
msg = amqp.Message(msg)
msg.properties["delivery_mode"] = 2
chan.basic_publish(msg, exchange="sorting_room", routing_key="1111")
chan.close()
conn.close()
代碼基本都是在 csdn 那個博客里面弄下來的。稍微的修改了以下。
啟動時，先運行 consumer 消費者進程，它會先連接， 并創建 Queue和 Exchange ，然后一直等待隊列中的消息。
然后，啟動 publisher ，它會先連接，然后向指定 Exchange 交換機推送帶有特定 routing_key 路由鍵的消息。
如果消費者對應的 Queue 隊列與 Exchange 交換機 的 routing_key 路由鍵 相對應的話。那么消費者就會接收到相應消息。至此，整個傳遞過程結束。
三、補充
注釋代碼
# -*- coding: utf-8 -*-
__author__ = 'lpe234'
__date__ = '2014-12-15'
import amqp
"""
amqp rabbitmq DEMO測試
先啟動 amqp_consumer.py 消費者，創建
"""
conn = amqp.Connection(host='localhost:5672', userid='guest', password='guest', virtual_host='/', insist=False)
# 每個channel都被分配了一個整數標識，自動由Connection()類的.channel()方法維護。可以使用.channel(x)來指定channel標識。
chan = conn.channel(channel_id=1)
# 當多個 channel_id 相同時，實際為同一 channel
# 現在已經有了一個可用的連接和channel。
# 現在將代碼分為兩類，生產者(producer)和消費者(consumer)。
# 創建一個消費者程序，會創建一個"po_box"的隊列和一個叫"sorting_room"的交換機。
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)
# 創建了"po_box" 的隊列，durable重啟之后會重新建立，auto_delete=False最后一個消費者斷開之后不會自動刪除，exclusive私有隊列
# 創建了"sorting_room"的交換機，type指定交換機類型，
# 現在已經有了一個可以接收消息的隊列和一個可以發送消息的交換機。不過還需要創建一個綁定
chan.queue_bind(queue='po_box', exchange='sorting_room', routing_key='jason')
# 這個綁定非常直接，任何送到交換機"sorting_room"的具有路由鍵"jason"的消息都被路由到"po_box" 隊列
# 現在有兩個方法，從隊列中取出消息。
# 第一個是調用 chan.basic_get()， 主動從隊列中拉出下一條消息(若沒有則返回 None)
# msg = chan.basic_get(queue='po_box')
# if msg:
# print msg.body
# chan.basic_ack(msg.delivery_tag)
# 第二種
def receive_callback(msg):
print msg.body
chan.basic_consume(queue='po_box', no_ack=True, callback=receive_callback, consumer_tag='testtag')
while True:
chan.wait()
chan.basic_cancel('testtag')
# chan.wait() 放在無限循環里面，這個函數會等待在隊列上，知道下一個消息到達隊列。
# chan.basic_cancel() 用來注銷該回調函數
# no_ack 這個參數，可以傳給 chan.basic_get(), chan.basic_consume。是否等待回饋，