python之消息隊列

 

引言

 

你是否遇到過兩個（多個）系統間需要通過定時任務來同步某些數據？你是否在為異構系統的不同進程間相互調用、通訊的問題而苦惱、掙扎？如果是，那么恭喜你，消息服務讓你可以很輕松地解決這些問題。
消息服務擅長於解決多系統、異構系統間的數據交換（消息通知/通訊）問題，你也可以把它用於系統間服務的相互調用（RPC）。本文將要介紹的RabbitMQ就是當前最主流的消息中間件之一。

 

 

RabbitMQ簡介

 

RabbitMQ是一個由erlang開發的AMQP（Advanced Message Queue ）的開源實現。AMQP 的出現其實也是應了廣大人民群眾的需求，雖然在同步消息通訊的世界里有很多公開標准（如 COBAR的 IIOP ，或者是 SOAP 等），但是在異步消息處理中卻不是這樣，只有大企業有一些商業實現（如微軟的 MSMQ ，IBM 的 Websphere MQ 等），因此，在 2006 年的 6 月，Cisco 、Redhat、iMatix 等聯合制定了 AMQP 的公開標准。

    RabbitMQ是由RabbitMQ Technologies Ltd開發並且提供商業支持的。該公司在2010年4月被SpringSource（VMWare的一個部門）收購。在2013年5月被並入Pivotal。其實VMWare，Pivotal和EMC本質上是一家的。不同的是VMWare是獨立上市子公司，而Pivotal是整合了EMC的某些資源，現在並沒有上市。

    RabbitMQ的官網是http://www.rabbitmq.com

 

        MQ全稱為Message Queue, 消息隊列（MQ）是一種應用程序對應用程序的通信方法。應用程序通過讀寫出入隊列的消息（針對應用程序的數據）來通信，而無需專用連接來鏈接它們。消 息傳遞指的是程序之間通過在消息中發送數據進行通信，而不是通過直接調用彼此來通信，直接調用通常是用於諸如遠程過程調用的技術。排隊指的是應用程序通過 隊列來通信。隊列的使用除去了接收和發送應用程序同時執行的要求。

MQ是消費-生產者模型的一個典型的代表，一端往消息隊列中不斷寫入消息，而另一端則可以讀取或者訂閱隊列中的消息。MQ和JMS類似，但不同的是JMS是SUN JAVA消息中間件服務的一個標准和API定義，而MQ則是遵循了AMQP協議的具體實現和產品。

 

應用場景的系統架構
 

 

• RabbitMQ Server：也叫broker server，它不是運送食物的卡車，而是一種傳輸服務。原話是RabbitMQ isn’t a food truck, it’s a delivery service. 他的角色就是維護一條從Producer到Consumer的路線，保證數據能夠按照指定的方式進行傳輸。但是這個保證也不是100%的保證，但是對於普通的應用來說這已經足夠了。當然對於商業系統來說，可以再做一層數據一致性的guard，就可以徹底保證系統的一致性了。
• Client A & B：也叫Producer，數據的發送方。Create messages and Publish (Send) them to a broker server (RabbitMQ)。一個Message有兩個部分：Payload（有效載荷）和Label（標簽）。Payload顧名思義就是傳輸的數據，Label是Exchange的名字或者說是一個tag，它描述了payload，而且RabbitMQ也是通過這個label來決定把這個Message發給哪個Consumer。AMQP僅僅描述了label，而RabbitMQ決定了如何使用這個label的規則。
• Client 1，2，3：也叫Consumer，數據的接收方。Consumers attach to a broker server (RabbitMQ) and subscribe to a queue。把queue比作是一個有名字的郵箱。當有Message到達某個郵箱后，RabbitMQ把它發送給它的某個訂閱者即Consumer。當然可能會把同一個Message發送給很多的Consumer。在這個Message中，只有payload，label已經被刪掉了。對於Consumer來說，它是不知道誰發送的這個信息的。就是協議本身不支持。但是當然了如果Producer發送的payload包含了Producer的信息就另當別論了。

對於一個數據從Producer到Consumer的正確傳遞，還有三個概念需要明確：exchanges, queues and bindings。

• Exchanges are where producers publish their messages. 消息交換機，它指定消息按什么規則，路由到哪個隊列
• Queues are where the messages end up and are received by consumers. 消息隊列載體，每個消息都會被投入到一個或多個隊列
• Bindings are how the messages get routed from the exchange to particular queues. 綁定，它的作用就是把exchange和queue按照路由規則綁定起來
• Routing Key：路由關鍵字，exchange根據這個關鍵字進行消息投遞

還有幾個概念是上述圖中沒有標明的，那就是Connection（連接），Channel（通道，頻道），Vhost（虛擬主機）。

• Connection：就是一個TCP的連接。Producer和Consumer都是通過TCP連接到RabbitMQ Server的。以后我們可以看到，程序的起始處就是建立這個TCP連接。
• Channel：虛擬連接。它建立在上述的TCP連接中。數據流動都是在Channel中進行的。也就是說，一般情況是程序起始建立TCP連接，第二步就是建立這個Channel。
• Vhost：虛擬主機，一個broker里可以開設多個vhost，用作不同用戶的權限分離。每個virtual host本質上都是一個RabbitMQ Server，擁有它自己的queue，exchagne，和bings rule等等。這保證了你可以在多個不同的application中使用RabbitMQ。

Channel的選擇

那么，為什么使用Channel，而不是直接使用TCP連接？

對於OS來說，建立和關閉TCP連接是有代價的，頻繁的建立關閉TCP連接對於系統的性能有很大的影響，而且TCP的連接數也有限制，這也限制了系統處理高並發的能力。但是，在TCP連接中建立Channel是沒有上述代價的。對於Producer或者Consumer來說，可以並發的使用多個Channel進行Publish或者Receive。
有實驗表明，1s的數據可以Publish10K的數據包。當然對於不同的硬件環境，不同的數據包大小這個數據肯定不一樣，但是我只想說明，對於普通的Consumer或者Producer來說，這已經足夠了。如果不夠用，你考慮的應該是如何細化split你的設計。

消息隊列執行過程

1. 客戶端連接到消息隊列服務器，打開一個Channel。
2. 客戶端聲明一個Exchange，並設置相關屬性。
3. 客戶端聲明一個Queue，並設置相關屬性。
4. 客戶端使用Routing key，在Exchange和Queue之間建立好綁定關系。
5. 客戶端投遞消息到Exchange。

Exchange接收到消息后，就根據消息的key和已經設置的Binding，進行消息路由，將消息投遞到一個或多個隊列里。有三種類型的Exchanges：direct，fanout，topic，每個實現了不同的路由算法（routing algorithm）：

• Direct exchange：完全根據key進行投遞的叫做Direct交換機。如果Routing key匹配, 那么Message就會被傳遞到相應的queue中。其實在queue創建時，它會自動的以queue的名字作為routing key來綁定那個exchange。例如，綁定時設置了Routing key為”abc”，那么客戶端提交的消息，只有設置了key為”abc”的才會投遞到隊列。
• Fanout exchange：不需要key的叫做Fanout交換機。它采取廣播模式，一個消息進來時，投遞到與該交換機綁定的所有隊列。
• Topic exchange：對key進行模式匹配后進行投遞的叫做Topic交換機。比如符號”#”匹配一個或多個詞，符號””匹配正好一個詞。例如”abc.#”匹配”abc.def.ghi”，”abc.”只匹配”abc.def”。

RabbitMQ是AMQP協議的實現。它主要包括以下組件：

1.Server(broker): 接受客戶端連接，實現AMQP消息隊列和路由功能的進程。

2.Virtual Host:其實是一個虛擬概念，類似於權限控制組，一個Virtual Host里面可以有若干個Exchange和Queue，但是權限控制的最小粒度是Virtual Host

3.Exchange:接受生產者發送的消息，並根據Binding規則將消息路由給服務器中的隊列。ExchangeType決定了Exchange路由消息的行為，例如，在RabbitMQ中，ExchangeType有direct、Fanout和Topic三種，不同類型的Exchange路由的行為是不一樣的。

4.Message Queue：消息隊列，用於存儲還未被消費者消費的消息。

5.Message: 由Header和Body組成，Header是由生產者添加的各種屬性的集合，包括Message是否被持久化、由哪個Message Queue接受、優先級是多少等。而Body是真正需要傳輸的APP數據。

6.Binding:Binding聯系了Exchange與Message Queue。Exchange在與多個Message Queue發生Binding后會生成一張路由表，路由表中存儲着Message Queue所需消息的限制條件即Binding Key。當Exchange收到Message時會解析其Header得到Routing Key，Exchange根據Routing Key與Exchange Type將Message路由到Message Queue。Binding Key由Consumer在Binding Exchange與Message Queue時指定，而Routing Key由Producer發送Message時指定，兩者的匹配方式由Exchange Type決定。 

7.Connection:連接，對於RabbitMQ而言，其實就是一個位於客戶端和Broker之間的TCP連接。

8.Channel:信道，僅僅創建了客戶端到Broker之間的連接后，客戶端還是不能發送消息的。需要為每一個Connection創建Channel，AMQP協議規定只有通過Channel才能執行AMQP的命令。一個Connection可以包含多個Channel。之所以需要Channel，是因為TCP連接的建立和釋放都是十分昂貴的，如果一個客戶端每一個線程都需要與Broker交互，如果每一個線程都建立一個TCP連接，暫且不考慮TCP連接是否浪費，就算操作系統也無法承受每秒建立如此多的TCP連接。RabbitMQ建議客戶端線程之間不要共用Channel，至少要保證共用Channel的線程發送消息必須是串行的，但是建議盡量共用Connection。

9.Command:AMQP的命令，客戶端通過Command完成與AMQP服務器的交互來實現自身的邏輯。例如在RabbitMQ中，客戶端可以通過publish命令發送消息，txSelect開啟一個事務，txCommit提交一個事務。

應用場景  

RabbitMQ，或者說AMQP解決了什么問題，或者說它的應用場景是什么？

 

     對於一個大型的軟件系統來說，它會有很多的組件或者說模塊或者說子系統或者（subsystem or Component or submodule）。那么這些模塊的如何通信？這和傳統的IPC有很大的區別。傳統的IPC很多都是在單一系統上的，模塊耦合性很大，不適合擴展（Scalability）；如果使用socket那么不同的模塊的確可以部署到不同的機器上，但是還是有很多問題需要解決。比如：

 

 1）信息的發送者和接收者如何維持這個連接，如果一方的連接中斷，這期間的數據如何防止丟失？

 2）如何降低發送者和接收者的耦合度？

 3）如何讓Priority高的接收者先接到數據？

 4）如何做到load balance？有效均衡接收者的負載？

 5）如何有效的將數據發送到相關的接收者？也就是說將接收者subscribe 不同的數據，如何做有效的filter。

 6）如何做到可擴展，甚至將這個通信模塊發到cluster上？

 7）如何保證接收者接收到了完整，正確的數據？

 

  AMDQ協議解決了以上的問題，而RabbitMQ實現了AMQP。

 

消息隊列的使用場景大概有3種：

1、系統集成，分布式系統的設計。各種子系統通過消息來對接，這種解決方案也逐步發展成一種架構風格，即“通過消息傳遞的架構”。

2、當系統中的同步處理方式嚴重影響了吞吐量，比如日志記錄。假如需要記錄系統中所有的用戶行為日志，如果通過同步的方式記錄日志勢必會影響系統的響應速度，當我們將日志消息發送到消息隊列，記錄日志的子系統就會通過異步的方式去消費日志消息。

3、系統的高可用性，比如電商的秒殺場景。當某一時刻應用服務器或數據庫服務器收到大量請求，將會出現系統宕機。如果能夠將請求轉發到消息隊列，再由服務器去消費這些消息將會使得請求變得平穩，提高系統的可用性。

 學習RabbitMQ的使用場景，來自官方教程：https://www.rabbitmq.com/getstarted.html

 

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

AMQP協議是一種二進制協議，提供客戶端應用與消息中間件之間異步、安全、高效地交互。從整體來看，AMQP協議可划分為三層。

 

這種分層架構類似於OSI網絡協議，可替換各層實現而不影響與其它層的交互。AMQP定義了合適的服務器端域模型，用於規范服務器的行為(AMQP服務器端可稱為broker)。

 

Model層決定這些基本域模型所產生的行為，這種行為在AMQP中用”command”表示，在后文中會着重來分析這些域模型。

 

Session層定義客戶端與broker之間的通信(通信雙方都是一個peer，可互稱做partner)，為command的可靠傳輸提供保障。

 

Transport層專注於數據傳送，並與Session保持交互，接受上層的數據，組裝成二進制流，傳送到receiver后再解析數據，交付給Session層。Session層需要Transport層完成網絡異常情況的匯報，順序傳送command等工作。

 

AMQP當中有四個概念非常重要：虛擬主機（virtual host），交換機（exchange），隊列（queue）和綁定（binding）。

 

虛擬主機（virtual host）：一個虛擬主機持有一組交換機、隊列和綁定。為什么需要多個虛擬主機呢？RabbitMQ當中，用戶只能在虛擬主機的粒度進行權限控制。因此，如果需要禁止A組訪問B組的交換機/隊列/綁定，必須為A和B分別創建一個虛擬主機。每一個RabbitMQ服務器都有一個默認的虛擬主機“/”。

 

隊列（Queue）：由消費者建立的，是messages的終點，可以理解成裝消息的容器。消息一直存在隊列里，直到有客戶端或者稱為Consumer消費者連接到這個隊列並將message取走為止。隊列可以有多個。

 

交換機（Exchange）：可以理解成具有路由表的路由程序。每個消息都有一個路由鍵（routing key），就是一個簡單的字符串。交換機中有一系列的綁定（binding），即路由規則（routes）。交換機可以有多個。多個隊列可以和同一個交換機綁定，同時多個交換機也可以和同一個隊列綁定。（多對多的關系）

 

三種交換機：

 

1.  Fanout Exchange（不處理路由鍵）：一個發送到交換機上的消息都會被轉發到與該交換機綁定的所有隊列上。Fanout交換機發消息是最快的。

 

2.  Direct Exchange（處理路由鍵）：如果一個隊列綁定到該交換機上，並且當前要求路由鍵為X，只有路由鍵是X的消息才會被這個隊列轉發。

 

3.  Topic Exchange（將路由鍵和某模式進行匹配，可以理解成模糊處理）：路由鍵的詞由“.”隔開，符號“#”表示匹配0個或多個詞，符號“*”表示匹配不多不少一個詞。因此“audit.#”能夠匹配到“audit.irs.corporate”，但是“audit.*”只會匹配到“audit.irs”

 

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

 

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

 

當你將消息發布到交換機的時候，可以指定一個標志“Delivery Mode”（投遞模式）。根據你使用的AMQP的庫不同，指定這個標志的方法可能不太一樣。簡單的說，就是將 Delivery Mode設置成2，也就是持久的即可。一般的AMQP庫都是將Delivery Mode設置成1，也就是非持久的。所以要持久化消息的步驟如下：

 

1.  將交換機設成 durable。

 

2.  將隊列設成 durable。

 

3.  將消息的 Delivery Mode 設置成2。

 

綁定（Bindings）如何持久化？我們無法在創建綁定的時候設置成durable。沒問題，如果綁定了一個 durable的隊列和一個durable的交換機，RabbitMQ會自動保留這個綁定。類似的，如果刪除了某個隊列或交換機（無論是不是 durable），依賴它的綁定都會自動刪除。

 

注意兩點：

 

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

 

2.  一旦創建了隊列和交換機，就不能修改其標志了。例如，如果創建了一個non-durable的隊列，然后想把它改變成durable的，唯一的辦法就是刪除這個隊列然后重現創建。因此，最好仔細檢查創建的標志。

 

消息隊列（MQ）使用過程

 

幾個概念說明：

 

1.  Broker：簡單來說就是消息隊列服務器實體。

 

2.  Exchange：消息交換機，它指定消息按什么規則，路由到哪個隊列。

 

3.  Queue：消息隊列載體，每個消息都會被投入到一個或多個隊列。

 

4.  Binding：綁定，它的作用就是把exchange和queue按照路由規則綁定起來。

 

5.  Routing Key：路由關鍵字，exchange根據這個關鍵字進行消息投遞。

 

6.  vhost：虛擬主機，一個broker里可以開設多個vhost，用作不同用戶的權限分離。

 

7.  producer：消息生產者，就是投遞消息的程序。

 

8.  consumer：消息消費者，就是接受消息的程序。

 

9.  channel：消息通道，在客戶端的每個連接里，可建立多個channel，每個channel代表一個會話任務。

 

消息隊列的使用過程大概如下：

 

1.  客戶端連接到消息隊列服務器，打開一個channel。

 

2.  客戶端聲明一個exchange，並設置相關屬性。

 

3.  客戶端聲明一個queue，並設置相關屬性。

 

4.  客戶端使用routing key，在exchange和queue之間建立好綁定關系。

 

5.  客戶端投遞消息到exchange。

 

6.  exchange接收到消息后，就根據消息的key和已經設置的binding，進行消息路由，將消息投遞到一個或多個隊列里。

 

rabbitMQ的優點（適用范圍）

 

1.        基於erlang語言開發具有高可用高並發的優點，適合集群服務器。

 

2.        健壯、穩定、易用、跨平台、支持多種語言、文檔齊全。

 

3.        有消息確認機制和持久化機制，可靠性高。

 

4.        開源

 

其他MQ的優勢：

 

1.        Apache ActiveMQ曝光率最高，但是可能會丟消息。

 

2.        ZeroMQ延遲很低、支持靈活拓撲，但是不支持消息持久化和崩潰恢復。

 

消息中間件主要用於組件之間的解耦，消息的發送者無需知道消息使用者的存在，反之亦然：

 

單向解耦

 

雙向解耦（如：RPC）

    例如一個日志系統，很容易使用RabbitMQ簡化工作量，一個Consumer可以進行消息的正常處理，另一個Consumer負責對消息進行日志記錄，只要在程序中指定兩個Consumer所監聽的queue以相同的方式綁定到同一個exchange即可，剩下的消息分發工作由RabbitMQ完成。
 

 

使用RabbitMQ server需要：

1. ErLang語言包；

2. RabbitMQ安裝包；

RabbitMQ同時提供了java的客戶端（一個jar包）。

 

安裝RabbitMQ

要想安裝RabbitMQ，首先需要安裝和配置好它的宿主環境erlang。去erlang官網下載好erlang otp_src源碼包，然后在本地執行源碼安裝。（erlang官網：http://www.erlang.org/）

RabbitMQ安裝（linux--服務端）

基礎環境：
內核
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

安裝API（客戶端）

pip install pika
or
easy_install pika
or
源碼
 
https://pypi.python.org/pypi/pika

 

用戶管理

1.  添加用戶：rabbitmqctl add_user username password

2.  刪除用戶：rabbitmqctl delete_user username

3.  修改密碼：rabbitmqctl change_password username newpassword

4.  清除密碼：rabbitmqctl clear_password {username}

5.  設置用戶標簽：rabbitmqctl set_user_tags {username} {tag…}如果tag為空則表示清除這個用戶的所有標簽

6.  列出所有用戶 ：rabbitmqctl list_users

權限控制

1.  創建虛擬主機：rabbitmqctl add_vhost vhostpath

2.  刪除虛擬主機：rabbitmqctl delete_vhost vhostpath

3.  列出所有虛擬主機：rabbitmqctl list_vhosts

4.  設置用戶權限：rabbitmqctl set_permissions [-p vhostpath] {username} {conf} {write} {read}

5.  清除用戶權限：rabbitmqctl clear_permissions [-p vhostpath] {username}

6.  列出虛擬主機上的所有權限：rabbitmqctl list_permissions [-p vhostpath]

7.  列出用戶權限：rabbitmqctl list_user_permissions {username}

 

RabbitMQ基本示例

 

實現最簡單的隊列通信

produce

 

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters("192.168.244.130",15672))
channel = connection.channel()
#聲明queue
channel.queue_declare(queue='hello')
channel.basic_publish(exchange="",
                      routing_key='hello',
                      body = 'Hello World!')
print("[x] Sent 'Hello World!")
connection.close()

 

 consume

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters("192.168.16.23"))
channel = connection.channel()
channel.queue_declare(queue="holle",durable=True)
def callback(ch,method,properties,body):
    print(ch,method,properties)
    print("[x] Received %r" %body)
    ch.basic_ack(delivery_tag=method.delivery_tag)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
                      queue="holle",
                      no_ack=True)


print('[*] waiting for messages. to exit press ctrl+c')
channel.start_consuming()

 

最基本的生產消費者模型

• 生產者代碼

#!/usr/bin/env python 3
import pika
#########  生產者 #########
#鏈接rabbit服務器（localhost是本機，如果是其他服務器請修改為ip地址）
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
#創建頻道
channel = connection.channel()
#創建一個隊列名叫test
channel.queue_declare(queue='test')
 
# channel.basic_publish向隊列中發送信息
# exchange -- 它使我們能夠確切地指定消息應該到哪個隊列去。
# routing_key 指定向哪個隊列中發送消息
# body是要插入的內容, 字符串格式
 
while True:  # 循環向隊列中發送信息，quit退出程序
    inp = input(">>>").strip()
    if inp == 'quit':
        break
    channel.basic_publish(exchange='',  
                          routing_key='test',
                          body=inp)
    print("生產者向隊列發送信息%s" % inp)
 
#緩沖區已經flush而且消息已經確認發送到了RabbitMQ中，關閉鏈接
connection.close()
 
# 輸出結果
>>>python
生產者向隊列發送信息python
>>>quit

消費者代碼

 

#!/usr/bin/env python 3
import pika
######### 消費者 #########
# 鏈接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
# 創建頻道
channel = connection.channel()
# 如果生產者沒有運行創建隊列，那么消費者也許就找不到隊列了。為了避免這個問題，所有消費者也創建這個隊列，如果隊列已經存在，則這條無效
channel.queue_declare(queue='test')
#接收消息需要使用callback這個函數來接收，他會被pika庫來調用,接受到的數據都是字節類型的
def callback(ch, method, properties, body):
     """
         ch : 代表 channel
         method ：隊列名
         properties : 連接rabbitmq時設置的屬性
         body : 從隊列中取到的內容，獲取到的數據時字節類型
     """
    print(" [x] Received %r" % body)
# channel.basic_consume 表示從隊列中取數據，如果拿到數據 那么將執行callback函數，callback是回調函數
# no_ack=True 表示消費完這個消息以后不主動把完成狀態通知rabbitmq
channel.basic_consume(callback,
                      queue='test',
                      no_ack=True)
print(' [*] 等待信息. To exit press CTRL+C')
#永遠循環等待數據處理和callback處理的數據，start_consuming方法會阻塞循環執行
channel.start_consuming()
 
# 輸出結果，一直等待處理隊列中的消息，不知終止，除非人為ctrl+c
 [*]等待消息，To exit press CTRL+C
 [x] Received b'python'

 

備注說明：

     生產者和消費者都連接到RabbitMQ Server 上，都會創建一個同名的queue，生產者向隊里中發送一條信息，消費者從隊列中獲取信息則完成通信。

如果生產者先啟動，則會先發送信息到隊列中，消費者啟動會直接會在隊列中取到生產者發送的信息內容。

如果消費者先啟動，則會阻塞住，一直等待生產者向隊列發送信息。

生產者發送一條信息后就結束了任務，而消費者一直在等待獲取新的信息。

 

消息持久化　

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()

消息持久化存儲（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兩者之間有一個持久化，一個非持久化，就不允許建立綁定。

 

消息公平分發

如果Rabbit只管按順序把消息發到各個消費者身上，不考慮消費者負載的話，很可能出現，一個機器配置不高的消費者那里堆積了很多消息處理不完，同時配置高的消費者卻一直很輕松。為解決此問題，可以在各個消費者端，配置perfetch=1,意思就是告訴RabbitMQ在我這個消費者當前消息還沒處理完的時候就不要再給我發新消息了。

 

channel.basic_qos(prefetch_count=1)

 

帶消息持久化+公平分發的完整代碼

生產者端

#!/usr/bin/env python
import pika
import sys
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()
 
channel.queue_declare(queue='task_queue', durable=True)
 
message = ' '.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(exchange='',
                      routing_key='task_queue',
                      body=message,
                      properties=pika.BasicProperties(
                         delivery_mode = 2, # make message persistent
                      ))
print(" [x] Sent %r" % message)
connection.close()

 消費者端

#!/usr/bin/env python
import pika
import time
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()
 
channel.queue_declare(queue='task_queue', durable=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
 
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    time.sleep(body.count(b'.'))
    print(" [x] Done")
    ch.basic_ack(delivery_tag = method.delivery_tag)
 
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
                      queue='task_queue')
 
channel.start_consuming()

 

Publish\Subscribe(消息發布\訂閱)　

RabbitMQ的發布與訂閱，借助於交換機(Exchange)來實現。

　　交換機的工作原理：消息發送端先將消息發送給交換機，交換機再將消息發送到綁定的消息隊列，而后每個接收端(consumer)都能從各自的消息隊列里接收到信息。

Exchange有三種工作模式，分別為：Fanout, Direct, Topic

 

• fanout : 所有bind到此exchange的queue都可以接受消息
• direct : 通過routingkey和exchange決定的那個唯一的queue可以接受消息
• topic : 所有符合routingkey（一個表達式）的routingkey所bind的queue

 當我們向隊列里發送消息時，其實並不是自己直接放入隊列中的，而是先交給exchange,然后由exchange放入指定的隊列中。想象下當我們要將一條消息發送到多個隊列中的時候，如果沒有exchange，我們需要發送多條到不同的隊列中，但是如果有了exchange，它會先和目標隊列建立一種綁定關系，當我們把一條消息發送到exchange中的時候，exchange會根據之前和隊列的綁定關系，將這條消息發送到所有與它有綁定關系的隊里中。

 

發布訂閱和簡單的消息隊列區別在於，發布訂閱會將消息發送給所有的訂閱者，而消息隊列中的數據被消費一次便消失了。所以，RabbitMQ實現發布和訂閱時，會為每一個訂閱者創建一個隊列，二發布者發布消息時，會將消息放置在所有相關隊列中。發布訂閱本質上就是發布端將消息發送給了exchange，exchange將消息發送給與它綁定關系的所有隊列。

exchange type = fanout   和exchange綁定關系的所有隊列都會收到信息

模式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使用正則模式即可。

消費者代碼

#!/usr/bin/env python3
#coding:utf8
import pika
import sys
 
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
 
channel.exchange_declare(exchange='topic_logs',type='topic')
 
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
 
binding_keys = sys.argv[1:]
if not binding_keys:
    sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
    sys.exit(1)
 
for binding_key in binding_keys:
    channel.queue_bind(exchange='topic_logs',
                       queue=queue_name,
                       routing_key=binding_key)
 
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 python3
#coding:utf8
import pika
import sys
 
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
 
channel.exchange_declare(exchange='topic_logs', type='topic')
 
routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='topic_logs',
                      routing_key=routing_key,
                      body=message)
print(" [x] Sent %r:%r" % (routing_key, message))
connection.close()

 

有選擇的接收消息(exchange type=direct)　

 

RabbitMQ還支持根據關鍵字發送，即：隊列綁定關鍵字，發送者將數據根據關鍵字發送到消息exchange，exchange根據 關鍵字 判定應該將數據發送至指定隊列。

 

publisher

 

import pika
import sys
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()
 
channel.exchange_declare(exchange='direct_logs',
                         type='direct')
 
severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='direct_logs',
                      routing_key=severity,
                      body=message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()

 subscriber

import pika
import sys
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()
 
channel.exchange_declare(exchange='direct_logs',
                         type='direct')
 
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
 
severities = sys.argv[1:]
if not severities:
    sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
    sys.exit(1)
 
for severity in severities:
    channel.queue_bind(exchange='direct_logs',
                       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()

 

 

 

更細致的消息過濾

 

Although using the direct exchange improved our system, it still has limitations - it can't do routing based on multiple criteria.

In our logging system we might want to subscribe to not only logs based on severity, but also based on the source which emitted the log. You might know this concept from the syslog unix tool, which routes logs based on both severity (info/warn/crit...) and facility (auth/cron/kern...).

That would give us a lot of flexibility - we may want to listen to just critical errors coming from 'cron' but also all logs from 'kern'.

publisher

import pika
import sys
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()
 
channel.exchange_declare(exchange='topic_logs',
                         type='topic')
 
routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='topic_logs',
                      routing_key=routing_key,
                      body=message)
print(" [x] Sent %r:%r" % (routing_key, message))
connection.close()

 subscriber

import pika
import sys
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()
 
channel.exchange_declare(exchange='topic_logs',
                         type='topic')
 
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
 
binding_keys = sys.argv[1:]
if not binding_keys:
    sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
    sys.exit(1)
 
for binding_key in binding_keys:
    channel.queue_bind(exchange='topic_logs',
                       queue=queue_name,
                       routing_key=binding_key)
 
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()

 

To receive all the logs run:

python receive_logs_topic.py "#"

To receive all logs from the facility "kern":

python receive_logs_topic.py "kern.*"

Or if you want to hear only about "critical" logs:

python receive_logs_topic.py "*.critical"

You can create multiple bindings:

python receive_logs_topic.py "kern.*" "*.critical" 

And to emit a log with a routing key "kern.critical" type:

python emit_log_topic.py "kern.critical" "A critical kernel error"

　　

Remote procedure call (RPC)

To illustrate how an RPC service could be used we're going to create a simple client class. It's going to expose a method named call which sends an RPC request and blocks until the answer is received:

1 2 3

fibonacci_rpc = FibonacciRpcClient() result = fibonacci_rpc.call( 4 ) print ( "fib(4) is %r" % result)

RPC server

#_*_coding:utf-8_*_
__author__ = 'Alex Li'
import pika
import time
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
 
channel = connection.channel()
 
channel.queue_declare(queue='rpc_queue')
 
def fib(n):
    if n == 0:
        return 0
    elif n == 1:
        return 1
    else:
        return fib(n-1) + fib(n-2)
 
def on_request(ch, method, props, body):
    n = int(body)
 
    print(" [.] fib(%s)" % n)
    response = fib(n)
 
    ch.basic_publish(exchange='',
                     routing_key=props.reply_to,
                     properties=pika.BasicProperties(correlation_id = \
                                                         props.correlation_id),
                     body=str(response))
    ch.basic_ack(delivery_tag = method.delivery_tag)
 
channel.basic_qos(prefetch_count=1)
channel.basic_consume(on_request, queue='rpc_queue')
 
print(" [x] Awaiting RPC requests")
channel.start_consuming()

 

RPC client

 

import pika
import uuid
 
class FibonacciRpcClient(object):
    def __init__(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(
                host='localhost'))
 
        self.channel = self.connection.channel()
 
        result = self.channel.queue_declare(exclusive=True)
        self.callback_queue = result.method.queue
 
        self.channel.basic_consume(self.on_response, no_ack=True,
                                   queue=self.callback_queue)
 
    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body
 
    def call(self, n):
        self.response = None
        self.corr_id = str(uuid.uuid4())
        self.channel.basic_publish(exchange='',
                                   routing_key='rpc_queue',
                                   properties=pika.BasicProperties(
                                         reply_to = self.callback_queue,
                                         correlation_id = self.corr_id,
                                         ),
                                   body=str(n))
        while self.response is None:
            self.connection.process_data_events()
        return int(self.response)
 
fibonacci_rpc = FibonacciRpcClient()
 
print(" [x] Requesting fib(30)")
response = fibonacci_rpc.call(30)
print(" [.] Got %r" % response)