介紹
RabbitMQ是一個在AMQP基礎上完整的，可復用的企業消息系統。他遵循Mozilla Public License開源協議。
MQ全稱為Message Queue, 消息隊列（MQ）是一種應用程序對應用程序的通信方法。應用程序通過讀寫出入隊列的消息（針對應用程序的數據）來通信，而無需專用連接來鏈接它們。消 息傳遞指的是程序之間通過在消息中發送數據進行通信，而不是通過直接調用彼此來通信，直接調用通常是用於諸如遠程過程調用的技術。排隊指的是應用程序通過 隊列來通信。隊列的使用除去了接收和發送應用程序同時執行的要求。
應用場景：
RabbitMQ無疑是目前最流行的消息隊列之一，對各種語言環境的支持也很豐富，作為一個.NET developer有必要學習和了解這一工具。消息隊列的使用場景大概有3種：
1、系統集成，分布式系統的設計。各種子系統通過消息來對接，這種解決方案也逐步發展成一種架構風格，即“通過消息傳遞的架構”。
2、當系統中的同步處理方式嚴重影響了吞吐量，比如日志記錄。假如需要記錄系統中所有的用戶行為日志，如果通過同步的方式記錄日志勢必會影響系統的響應速度，當我們將日志消息發送到消息隊列，記錄日志的子系統就會通過異步的方式去消費日志消息。
3、系統的高可用性，比如電商的秒殺場景。當某一時刻應用服務器或數據庫服務器收到大量請求，將會出現系統宕機。如果能夠將請求轉發到消息隊列，再由服務器去消費這些消息將會使得請求變得平穩，提高系統的可用性。
安裝
安裝RabbitMQ
基礎環境： 內核 3.10.0-327.el7.x86_64 系統版本 CentOS Linux release 7.2.1511 (Core) 安裝配置epel源 # rpm -ivh http://mirrors.neusoft.edu.cn/epel/7/x86_64/e/epel-release-7-7.noarch.rpm 安裝erlang # yum install erlang 下載RabbitMQ 3.6.1 # wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.1/rabbitmq-server-3.6.1-1.noarch.rpm 安裝rabbitmq-server # rpm -ivh rabbitmq-server-3.6.1-1.noarch.rpm 生成配置文件 # cp /usr/share/doc/rabbitmq-server-3.6.1/rabbitmq.config.example /etc/rabbitmq/rabbitmq.config 啟動RabbitMQ # rabbitmq-server start
安裝Python API
# pip3 install pika or # easy_install pika
Python 操作RabbitMQ
對於RabbitMQ來說，生產和消費不再針對內存里的一個Queue對象，而是某台服務器上的RabbitMQ Server實現的消息隊列。
1.生產者代碼
#!/usr/bin/env python # -*- coding: utf-8 -*- # auth : pangguoping import pika # ######################### 生產者 ######################### credentials = pika.PlainCredentials('admin', 'admin') #鏈接rabbit服務器（localhost是本機，如果是其他服務器請修改為ip地址） connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.103',5672,'/',credentials)) #創建頻道 channel = connection.channel() # 聲明消息隊列，消息將在這個隊列中進行傳遞。如果將消息發送到不存在的隊列，rabbitmq將會自動清除這些消息。如果隊列不存在，則創建 channel.queue_declare(queue='hello') #exchange -- 它使我們能夠確切地指定消息應該到哪個隊列去。 #向隊列插入數值 routing_key是隊列名 body是要插入的內容 channel.basic_publish(exchange='', routing_key='hello', body='Hello World!') print("開始隊列") #緩沖區已經flush而且消息已經確認發送到了RabbitMQ中，關閉鏈接 connection.close()
2.消費者代碼
#!/usr/bin/env python # -*- coding: utf-8 -*- # auth : pangguoping import pika # ########################## 消費者 ########################## credentials = pika.PlainCredentials('admin', 'admin') # 連接到rabbitmq服務器 connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.103',5672,'/',credentials)) channel = connection.channel() # 聲明消息隊列，消息將在這個隊列中進行傳遞。如果隊列不存在，則創建 channel.queue_declare(queue='wzg') # 定義一個回調函數來處理，這邊的回調函數就是將信息打印出來。 def callback(ch, method, properties, body): print(" [x] Received %r" % body) # 告訴rabbitmq使用callback來接收信息 channel.basic_consume(callback, queue='hello', no_ack=True) # no_ack=True表示在回調函數中不需要發送確認標識 print(' [*] Waiting for messages. To exit press CTRL+C') # 開始接收信息，並進入阻塞狀態，隊列里有信息才會調用callback進行處理。按ctrl+c退出。 channel.start_consuming()
RabbitMQ持久化
1、acknowledgment 消息不丟失的方法
生效方法：channel.basic_consume(consumer_callback, queue, no_ack=False, exclusive=False, consumer_tag=None, arguments=None)
即no_ack=False（默認為False，即必須有確認標識）,在回調函數consumer_callback中，未收到確認標識，那么，RabbitMQ會重新將該任務添加到隊列中。
生產者代碼
#!/usr/bin/env python # -*- coding: utf-8 -*- # auth : pangguoping import pika # ######################### 生產者 ######################### credentials = pika.PlainCredentials('admin', 'admin') #鏈接rabbit服務器（localhost是本機，如果是其他服務器請修改為ip地址） connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.103',5672,'/',credentials)) #創建頻道 channel = connection.channel() # 聲明消息隊列，消息將在這個隊列中進行傳遞。如果將消息發送到不存在的隊列，rabbitmq將會自動清除這些消息。如果隊列不存在，則創建 channel.queue_declare(queue='hello') #exchange -- 它使我們能夠確切地指定消息應該到哪個隊列去。 #向隊列插入數值 routing_key是隊列名 body是要插入的內容 channel.basic_publish(exchange='', routing_key='hello', body='Hello World!') print("開始隊列") #緩沖區已經flush而且消息已經確認發送到了RabbitMQ中，關閉鏈接 connection.close()
消費者代碼：
import pika credentials = pika.PlainCredentials('admin', 'admin') # 鏈接rabbit connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.103',5672,'/',credentials)) # 創建頻道 channel = connection.channel() # 如果生產者沒有運行創建隊列，那么消費者創建隊列 channel.queue_declare(queue='hello') def callback(ch, method, properties, body): print(" [x] Received %r" % body) import time time.sleep(10) print 'ok' ch.basic_ack(delivery_tag=method.delivery_tag) # 主要使用此代碼 channel.basic_consume(callback, queue='hello', no_ack=False) print(' [*] Waiting for messages. To exit press CTRL+C') channel.start_consuming()
2、消息持久化存儲（Message durability）
雖然有了消息反饋機制，但是如果rabbitmq自身掛掉的話，那么任務還是會丟失。所以需要將任務持久化存儲起來。聲明持久化存儲
channel.queue_declare(queue='wzg', durable=True) # 聲明隊列持久化
Ps: 但是這樣程序會執行錯誤，因為‘wzg’這個隊列已經存在，並且是非持久化的，rabbitmq不允許使用不同的參數來重新定義存在的隊列。因此需要重新定義一個隊列
channel.queue_declare(queue='test_queue', durable=True) # 聲明隊列持久化
注意：如果僅僅是設置了隊列的持久化，僅隊列本身可以在rabbit-server宕機后保留，隊列中的信息依然會丟失，如果想讓隊列中的信息或者任務保留，還需要做以下設置：
channel.basic_publish(exchange='', routing_key="test_queue", body=message, properties=pika.BasicProperties( delivery_mode = 2, # 使消息或任務也持久化存儲 ))
消息隊列持久化包括3個部分： （1）exchange持久化，在聲明時指定durable => 1 （2）queue持久化，在聲明時指定durable => 1 （3）消息持久化，在投遞時指定delivery_mode=> 2（1是非持久化） 如果exchange和queue都是持久化的，那么它們之間的binding也是持久化的。如果exchange和queue兩者之間有一個持久化，一個非持久化，就不允許建立綁定。
發布與訂閱
RabbitMQ的發布與訂閱，借助於交換機(Exchange)來實現。
交換機的工作原理：消息發送端先將消息發送給交換機，交換機再將消息發送到綁定的消息隊列，而后每個接收端(consumer)都能從各自的消息隊列里接收到信息。
Exchange有三種工作模式，分別為：Fanout, Direct, Topic
模式1 Fanout
任何發送到Fanout Exchange的消息都會被轉發到與該Exchange綁定(Binding)的所有Queue上
1.可以理解為路由表的模式
2.這種模式不需要routing_key(即使指定，也是無效的)
3.這種模式需要提前將Exchange與Queue進行綁定，一個Exchange可以綁定多個Queue，一個Queue可以同多個Exchange進行綁定。
4.如果接受到消息的Exchange沒有與任何Queue綁定，則消息會被拋棄。
注意：這個時候必須先啟動消費者，即訂閱者。因為隨機隊列是在consumer啟動的時候隨機生成的，並且進行綁定的。producer僅僅是發送至exchange，並不直接與隨機隊列進行通信。
生產者代碼
#!/usr/bin/env python # -*- coding: utf-8 -*- # auth : pangguoping # rabbitmq 發布者 import pika credentials = pika.PlainCredentials('admin', 'admin') #鏈接rabbit服務器（localhost是本機，如果是其他服務器請修改為ip地址） connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.103',5672,'/',credentials)) channel = connection.channel() # 定義交換機，exchange表示交換機名稱，type表示類型 channel.exchange_declare(exchange='logs_fanout', type='fanout') message = 'Hello Python' # 將消息發送到交換機 channel.basic_publish(exchange='logs_fanout', # 指定exchange routing_key='', # fanout下不需要配置，配置了也不會生效 body=message) print(" [x] Sent %r" % message) connection.close()
消費者代碼
#!/usr/bin/env python # -*- coding: utf-8 -*- # auth : pangguoping # rabbitmq 訂閱者 import pika credentials = pika.PlainCredentials('admin', 'admin') #鏈接rabbit服務器（localhost是本機，如果是其他服務器請修改為ip地址） connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.103',5672,'/',credentials)) channel = connection.channel() # 定義交換機，進行exchange聲明，exchange表示交換機名稱，type表示類型 channel.exchange_declare(exchange='logs_fanout', type='fanout') # 隨機創建隊列 result = channel.queue_declare(exclusive=True) # exclusive=True表示建立臨時隊列，當consumer關閉后，該隊列就會被刪除 queue_name = result.method.queue # 將隊列與exchange進行綁定 channel.queue_bind(exchange='logs_fanout', queue=queue_name) print(' [*] Waiting for logs. To exit press CTRL+C') def callback(ch, method, properties, body): print(" [x] %r" % body) # 從隊列獲取信息 channel.basic_consume(callback, queue=queue_name, no_ack=True) channel.start_consuming()
模式2 Direct
路由鍵的工作原理：每個接收端的消息隊列在綁定交換機的時候，可以設定相應的路由鍵。發送端通過交換機發送信息時，可以指明路由鍵 ，交換機會根據路由鍵把消息發送到相應的消息隊列，這樣接收端就能接收到消息了。
任何發送到Direct Exchange的消息都會被轉發到routing_key中指定的Queue：
1.一般情況可以使用rabbitMQ自帶的Exchange：”” (該Exchange的名字為空字符串)， 也可以自定義Exchange
2.這種模式下不需要將Exchange進行任何綁定(bind)操作。當然也可以進行綁定。可以將不同的routing_key與不同的queue進行綁定，不同的queue與不同exchange進行綁定
3.消息傳遞時需要一個“routing_key”
4.如果消息中中不存在routing_key中綁定的隊列名，則該消息會被拋棄。
如果一個exchange 聲明為direct，並且bind中指定了routing_key,那么發送消息時需要同時指明該exchange和routing_key.
消費者代碼
#!/usr/bin/env python # -*- coding: utf-8 -*- # auth : pangguoping # 消費者 import pika credentials = pika.PlainCredentials('admin', 'admin') #鏈接rabbit服務器（localhost是本機，如果是其他服務器請修改為ip地址） connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.103',5672,'/',credentials)) channel = connection.channel() # 定義exchange和類型 channel.exchange_declare(exchange='direct_test', type='direct') # 生成隨機隊列 result = channel.queue_declare(exclusive=True) queue_name = result.method.queue severities = ['error', ] # 將隨機隊列與routing_key關鍵字以及exchange進行綁定 for severity in severities: channel.queue_bind(exchange='direct_test', queue=queue_name, routing_key=severity) print(' [*] Waiting for logs. To exit press CTRL+C') def callback(ch, method, properties, body): print(" [x] %r:%r" % (method.routing_key, body)) # 接收消息 channel.basic_consume(callback, queue=queue_name, no_ack=True) channel.start_consuming()
生產者
#!/usr/bin/env python # -*- coding: utf-8 -*- # auth : pangguoping # 發布者 import pika credentials = pika.PlainCredentials('admin', 'admin') #鏈接rabbit服務器（localhost是本機，如果是其他服務器請修改為ip地址） connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.103',5672,'/',credentials)) channel = connection.channel() # 定義交換機名稱及類型 channel.exchange_declare(exchange='direct_test', type='direct') severity = 'info' message = '123' # 發布消息至交換機direct_test，且發布的消息攜帶的關鍵字routing_key是info channel.basic_publish(exchange='direct_test', routing_key=severity, body=message) print(" [x] Sent %r:%r" % (severity, message)) connection.close()
當接收端正在運行時，可以使用rabbitmqctl list_bindings來查看綁定情況。
模式3 Topic
路由鍵模糊匹配，其實是路由鍵(routing_key)的擴展，就是可以使用正則表達式，和常用的正則表示式不同，這里的話“#”表示所有、全部的意思；“*”只匹配到一個詞。
任何發送到Topic Exchange的消息都會被轉發到所有關心routing_key中指定話題的Queue上
1.這種模式較為復雜，簡單來說，就是每個隊列都有其關心的主題，所有的消息都帶有一個“標題”(routing_key)，Exchange會將消息轉發到所有關注主題能與 routing_key模糊匹配的隊列。
2.這種模式需要routing_key，也許要提前綁定Exchange與Queue。
3.在進行綁定時，要提供一個該隊列關心的主題，如“#.log.#”表示該隊列關心所有涉及log的消息(一個routing_key為”MQ.log.error”的消息會被轉發到該隊列)。
4.“#”表示0個或若干個關鍵字，“*”表示一個關鍵字。如“log.*”能與“log.warn”匹配，無法與“log.warn.timeout”匹配；但是“log.#”能與上述兩者匹配。
5.同樣，如果Exchange沒有發現能夠與routing_key匹配的Queue，則會拋棄此消息。
具體代碼這里不在多余寫，參照第二種模式的就可以，唯一變動的地方就是exchange type的聲明，以及進行綁定和發送的時候routing_key使用正則模式即可。
至此，利用Python操作RabbitMQ，以及一些簡單用法，在這里介紹完畢。
參考http://www.cnblogs.com/jishuweiwang