python之消息隊列

時間 2019-11-19

標籤 python 消息隊列欄目 Python 简体版

原文原文鏈接

引言

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

RabbitMQ簡介

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

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

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

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你的設計。安全

消息隊列執行過程

客戶端鏈接到消息隊列服務器，打開一個Channel。
客戶端聲明一個Exchange，並設置相關屬性。
客戶端聲明一個Queue，並設置相關屬性。
客戶端使用Routing key，在Exchange和Queue之間創建好綁定關係。
客戶端投遞消息到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種：

一、系統集成，分佈式系統的設計。各類子系統經過消息來對接，這種解決方案也逐步發展成一種架構風格，即「經過消息傳遞的架構」。

二、當系統中的同步處理方式嚴重影響了吞吐量，好比日誌記錄。假如須要記錄系統中全部的用戶行爲日誌，若是經過同步的方式記錄日誌勢必會影響系統的響應速度，當咱們將日誌消息發送到消息隊列，記錄日誌的子系統就會經過異步的方式去消費日誌消息。

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

學習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:

 
          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)

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。