RabbitMQ基礎概念詳細介紹 rabbitmq 實現原理

時間 2019-12-08

原文原文鏈接

http://www.diggerplus.org/archives/3110html

本文對rabbitmq基礎介紹，徹底是爲了下一篇rabbitmq性能測試作準備，讓讀者去了解咱們須要測試的是什麼樣一個「東西」。java

引言

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

RabbitMQ簡介

AMQP，即Advanced Message Queuing Protocol，高級消息隊列協議，是應用層協議的一個開放標準，爲面向消息的中間件設計。消息中間件主要用於組件之間的解耦，消息的發送者無需知道消息使用者的存在，反之亦然。
AMQP的主要特徵是面向消息、隊列、路由（包括點對點和發佈/訂閱）、可靠性、安全。
RabbitMQ是一個開源的AMQP實現，服務器端用Erlang語言編寫，支持多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用於在分佈式系統中存儲轉發消息，在易用性、擴展性、高可用性等方面表現不俗。
下面將重點介紹RabbitMQ中的一些基礎概念，瞭解了這些概念，是使用好RabbitMQ的基礎。web

ConnectionFactory、Connection、Channel

ConnectionFactory、Connection、Channel都是RabbitMQ對外提供的API中最基本的對象。Connection是RabbitMQ的socket連接，它封裝了socket協議相關部分邏輯。ConnectionFactory爲Connection的製造工廠。
Channel是咱們與RabbitMQ打交道的最重要的一個接口，咱們大部分的業務操做是在Channel這個接口中完成的，包括定義Queue、定義Exchange、綁定Queue與Exchange、發佈消息等。正則表達式

Queue

Queue（隊列）是RabbitMQ的內部對象，用於存儲消息，用下圖表示。
shell

RabbitMQ中的消息都只能存儲在Queue中，生產者（下圖中的P）生產消息並最終投遞到Queue中，消費者（下圖中的C）能夠從Queue中獲取消息並消費。瀏覽器

多個消費者能夠訂閱同一個Queue，這時Queue中的消息會被平均分攤給多個消費者進行處理，而不是每一個消費者都收到全部的消息並處理。
安全

Message acknowledgment

在實際應用中，可能會發生消費者收到Queue中的消息，但沒有處理完成就宕機（或出現其餘意外）的狀況，這種狀況下就可能會致使消息丟失。爲了不這種狀況發生，咱們能夠要求消費者在消費完消息後發送一個回執給RabbitMQ，RabbitMQ收到消息回執（Message acknowledgment）後纔將該消息從Queue中移除；若是RabbitMQ沒有收到回執並檢測到消費者的RabbitMQ鏈接斷開，則RabbitMQ會將該消息發送給其餘消費者（若是存在多個消費者）進行處理。這裏不存在timeout概念，一個消費者處理消息時間再長也不會致使該消息被髮送給其餘消費者，除非它的RabbitMQ鏈接斷開。
這裏會產生另一個問題，若是咱們的開發人員在處理完業務邏輯後，忘記發送回執給RabbitMQ，這將會致使嚴重的bug——Queue中堆積的消息會愈來愈多；消費者重啓後會重複消費這些消息並重復執行業務邏輯…服務器

Message durability

若是咱們但願即便在RabbitMQ服務重啓的狀況下，也不會丟失消息，咱們能夠將Queue與Message都設置爲可持久化的（durable），這樣能夠保證絕大部分狀況下咱們的RabbitMQ消息不會丟失。但依然解決不了小几率丟失事件的發生（好比RabbitMQ服務器已經接收到生產者的消息，但還沒來得及持久化該消息時RabbitMQ服務器就斷電了），若是咱們須要對這種小几率事件也要管理起來，那麼咱們要用到事務。因爲這裏僅爲RabbitMQ的簡單介紹，因此這裏將不講解RabbitMQ相關的事務。架構

Prefetch count

前面咱們講到若是有多個消費者同時訂閱同一個Queue中的消息，Queue中的消息會被平攤給多個消費者。這時若是每一個消息的處理時間不一樣，就有可能會致使某些消費者一直在忙，而另一些消費者很快就處理完手頭工做並一直空閒的狀況。咱們能夠經過設置prefetchCount來限制Queue每次發送給每一個消費者的消息數，好比咱們設置prefetchCount=1，則Queue每次給每一個消費者發送一條消息；消費者處理完這條消息後Queue會再給該消費者發送一條消息。

Exchange

在上一節咱們看到生產者將消息投遞到Queue中，實際上這在RabbitMQ中這種事情永遠都不會發生。實際的狀況是，生產者將消息發送到Exchange（交換器，下圖中的X），由Exchange將消息路由到一個或多個Queue中（或者丟棄）。

Exchange是按照什麼邏輯將消息路由到Queue的？這個將在Binding一節介紹。
RabbitMQ中的Exchange有四種類型，不一樣的類型有着不一樣的路由策略，這將在Exchange Types一節介紹。

routing key

生產者在將消息發送給Exchange的時候，通常會指定一個routing key，來指定這個消息的路由規則，而這個routing key須要與Exchange Type及binding key聯合使用才能最終生效。
在Exchange Type與binding key固定的狀況下（在正常使用時通常這些內容都是固定配置好的），咱們的生產者就能夠在發送消息給Exchange時，經過指定routing key來決定消息流向哪裏。
RabbitMQ爲routing key設定的長度限制爲255 bytes。

Binding

RabbitMQ中經過Binding將Exchange與Queue關聯起來，這樣RabbitMQ就知道如何正確地將消息路由到指定的Queue了。

Binding key

在綁定（Binding）Exchange與Queue的同時，通常會指定一個binding key；消費者將消息發送給Exchange時，通常會指定一個routing key；當binding key與routing key相匹配時，消息將會被路由到對應的Queue中。這個將在Exchange Types章節會列舉實際的例子加以說明。
在綁定多個Queue到同一個Exchange的時候，這些Binding容許使用相同的binding key。
binding key 並非在全部狀況下都生效，它依賴於Exchange Type，好比fanout類型的Exchange就會無視binding key，而是將消息路由到全部綁定到該Exchange的Queue。

Exchange Types

RabbitMQ經常使用的Exchange Type有fanout、direct、topic、headers這四種（AMQP規範裏還提到兩種Exchange Type，分別爲system與自定義，這裏不予以描述），下面分別進行介紹。

fanout

fanout類型的Exchange路由規則很是簡單，它會把全部發送到該Exchange的消息路由到全部與它綁定的Queue中。

上圖中，生產者（P）發送到Exchange（X）的全部消息都會路由到圖中的兩個Queue，並最終被兩個消費者（C1與C2）消費。

direct

direct類型的Exchange路由規則也很簡單，它會把消息路由到那些binding key與routing key徹底匹配的Queue中。

以上圖的配置爲例，咱們以routingKey=」error」發送消息到Exchange，則消息會路由到Queue1（amqp.gen-S9b…，這是由RabbitMQ自動生成的Queue名稱）和Queue2（amqp.gen-Agl…）；若是咱們以routingKey=」info」或routingKey=」warning」來發送消息，則消息只會路由到Queue2。若是咱們以其餘routingKey發送消息，則消息不會路由到這兩個Queue中。

topic

前面講到direct類型的Exchange路由規則是徹底匹配binding key與routing key，但這種嚴格的匹配方式在不少狀況下不能知足實際業務需求。topic類型的Exchange在匹配規則上進行了擴展，它與direct類型的Exchage類似，也是將消息路由到binding key與routing key相匹配的Queue中，但這裏的匹配規則有些不一樣，它約定：

routing key爲一個句點號「. 」分隔的字符串（咱們將被句點號「. 」分隔開的每一段獨立的字符串稱爲一個單詞），如「stock.usd.nyse」、「nyse.vmw」、「quick.orange.rabbit」
binding key與routing key同樣也是句點號「. 」分隔的字符串
binding key中能夠存在兩種特殊字符「*」與「#」，用於作模糊匹配，其中「*」用於匹配一個單詞，「#」用於匹配多個單詞（能夠是零個）

以上圖中的配置爲例，routingKey=」quick.orange.rabbit」的消息會同時路由到Q1與Q2，routingKey=」lazy.orange.fox」的消息會路由到Q1，routingKey=」lazy.brown.fox」的消息會路由到Q2，routingKey=」lazy.pink.rabbit」的消息會路由到Q2（只會投遞給Q2一次，雖然這個routingKey與Q2的兩個bindingKey都匹配）；routingKey=」quick.brown.fox」、routingKey=」orange」、routingKey=」quick.orange.male.rabbit」的消息將會被丟棄，由於它們沒有匹配任何bindingKey。

headers

headers類型的Exchange不依賴於routing key與binding key的匹配規則來路由消息，而是根據發送的消息內容中的headers屬性進行匹配。
在綁定Queue與Exchange時指定一組鍵值對；當消息發送到Exchange時，RabbitMQ會取到該消息的headers（也是一個鍵值對的形式），對比其中的鍵值對是否徹底匹配Queue與Exchange綁定時指定的鍵值對；若是徹底匹配則消息會路由到該Queue，不然不會路由到該Queue。
該類型的Exchange沒有用到過（不過也應該頗有用武之地），因此不作介紹。

RPC

MQ自己是基於異步的消息處理，前面的示例中全部的生產者（P）將消息發送到RabbitMQ後不會知道消費者（C）處理成功或者失敗（甚至連有沒有消費者來處理這條消息都不知道）。
但實際的應用場景中，咱們極可能須要一些同步處理，須要同步等待服務端將個人消息處理完成後再進行下一步處理。這至關於RPC（Remote Procedure Call，遠程過程調用）。在RabbitMQ中也支持RPC。

RabbitMQ中實現RPC的機制是：

客戶端發送請求（消息）時，在消息的屬性（MessageProperties，在AMQP協議中定義了14中properties，這些屬性會隨着消息一塊兒發送）中設置兩個值replyTo（一個Queue名稱，用於告訴服務器處理完成後將通知個人消息發送到這個Queue中）和correlationId（這次請求的標識號，服務器處理完成後須要將此屬性返還，客戶端將根據這個id瞭解哪條請求被成功執行了或執行失敗）
服務器端收到消息並處理
服務器端處理完消息後，將生成一條應答消息到replyTo指定的Queue，同時帶上correlationId屬性
客戶端以前已訂閱replyTo指定的Queue，從中收到服務器的應答消息後，根據其中的correlationId屬性分析哪條請求被執行了，根據執行結果進行後續業務處理

總結

本文介紹了RabbitMQ中我的認爲最重要的概念，充分利用RabbitMQ提供的這些功能就能夠處理咱們絕大部分的異步業務了。
本篇的基本概念可能很難理解並消化，結合實際的應用代碼應該會比較容易吸取。因此接下來要寫的文章例中會包含實際的業務應用場景分析，爲何使用RabbitMQ來實現，如何用RabbitMQ來實現。

http://rabbitmq-into-chinese.readthedocs.org/zh_CN/latest/tutorials_with_python/[6]RPC/#_5

http://blog.csdn.net/anzhsoft/article/details/19563091

一個客戶端能夠和多個exchange相連

consumer指定的時候只要指定QueueName就OK了

這個系統架構圖版權屬於sunjun041640。

RabbitMQ Server：也叫broker server，它不是運送食物的卡車，而是一種傳輸服務。原話是RabbitMQisn’t a food truck, it’s a delivery service. 他的角色就是維護一條從Producer到Consumer的路線，保證數據可以按照指定的方式進行傳輸。可是這個保證也不是100%的保證，可是對於普通的應用來講這已經足夠了。固然對於商業系統來講，能夠再作一層數據一致性的guard，就能夠完全保證系統的一致性了。

Client A & B：也叫Producer，數據的發送方。createmessages 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，數據的接收方。Consumersattach 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.

Queuesare where the messages end up and are received by consumers

Bindings are how the messages get routed from the exchange to particular queues.

還有幾個概念是上述圖中沒有標明的，那就是Connection（鏈接），Channel（通道，頻道）。

Connection：就是一個TCP的鏈接。Producer和Consumer都是經過TCP鏈接到RabbitMQ Server的。之後咱們能夠看到，程序的起始處就是創建這個TCP鏈接。

Channels： 虛擬鏈接。它創建在上述的TCP鏈接中。數據流動都是在Channel中進行的。也就是說，通常狀況是程序起始創建TCP鏈接，第二步就是創建這個Channel。

那麼，爲何使用Channel，而不是直接使用TCP鏈接？

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

對於API的使用能夠參考官方的java文檔，（裏面的思想不管是java、cpp仍是python都類似）

https://www.rabbitmq.com/releases/rabbitmq-java-client/v3.1.1/rabbitmq-java-client-javadoc-3.1.1/com/rabbitmq/client/Channel.html#basicConsume(java.lang.String, boolean, java.lang.String, com.rabbitmq.client.Consumer)

http://m.oschina.net/blog/186202

從AMQP協議能夠看出，MessageQueue、Exchange和Binding構成了AMQP協議的核心，下面咱們就圍繞這三個主要組件從應用使用的角度全面的介紹如何利用Rabbit MQ構建消息隊列以及使用過程當中的注意事項。

1. 聲明MessageQueue

在Rabbit MQ中，不管是生產者發送消息仍是消費者接受消息，都首先須要聲明一個MessageQueue。這就存在一個問題，是生產者聲明仍是消費者聲明呢？要解決這個問題，首先須要明確：

a)消費者是沒法訂閱或者獲取不存在的MessageQueue中信息。

b)消息被Exchange接受之後，若是沒有匹配的Queue，則會被丟棄。

在明白了上述兩點之後，就容易理解若是是消費者去聲明Queue，就有可能會出如今聲明Queue以前，生產者已發送的消息被丟棄的隱患。若是應用可以經過消息重發的機制容許消息丟失，則使用此方案沒有任何問題。可是若是不能接受該方案，這就須要不管是生產者仍是消費者，在發送或者接受消息前，都須要去嘗試創建消息隊列。這裏有一點須要明確，若是客戶端嘗試創建一個已經存在的消息隊列，Rabbit MQ不會作任何事情，並返回客戶端創建成功的。

若是一個消費者在一個信道中正在監聽某一個隊列的消息，Rabbit MQ是不容許該消費者在同一個channel去聲明其餘隊列的。Rabbit MQ中，能夠經過queue.declare命令聲明一個隊列，能夠設置該隊列如下屬性:

a) Exclusive：排他隊列，若是一個隊列被聲明爲排他隊列，該隊列僅對首次聲明它的鏈接可見，並在鏈接斷開時自動刪除。這裏須要注意三點：其一，排他隊列是基於鏈接可見的，同一鏈接的不一樣信道是能夠同時訪問同一個鏈接建立的排他隊列的。其二，「首次」，若是一個鏈接已經聲明瞭一個排他隊列，其餘鏈接是不容許創建同名的排他隊列的，這個與普通隊列不一樣。其三，即便該隊列是持久化的，一旦鏈接關閉或者客戶端退出，該排他隊列都會被自動刪除的。這種隊列適用於只限於一個客戶端發送讀取消息的應用場景。

b) Auto-delete:自動刪除，若是該隊列沒有任何訂閱的消費者的話，該隊列會被自動刪除。這種隊列適用於臨時隊列。

c) Durable:持久化，這個會在後面做爲專門一個章節討論。

d) 其餘選項，例如若是用戶僅僅想查詢某一個隊列是否已存在，若是不存在，不想創建該隊列，仍然能夠調用queue.declare，只不過須要將參數passive設爲true，傳給queue.declare，若是該隊列已存在，則會返回true；若是不存在，則會返回Error，可是不會建立新的隊列。

2. 生產者發送消息

在AMQP模型中，Exchange是接受生產者消息並將消息路由到消息隊列的關鍵組件。ExchangeType和Binding決定了消息的路由規則。因此生產者想要發送消息，首先必需要聲明一個Exchange和該Exchange對應的Binding。能夠經過 ExchangeDeclare和BindingDeclare完成。在Rabbit MQ中，聲明一個Exchange須要三個參數：ExchangeName，ExchangeType和Durable。ExchangeName是該Exchange的名字，該屬性在建立Binding和生產者經過publish推送消息時須要指定。ExchangeType，指Exchange的類型，在RabbitMQ中，有三種類型的Exchange：direct ，fanout和topic，不一樣的Exchange會表現出不一樣路由行爲。Durable是該Exchange的持久化屬性，這個會在消息持久化章節討論。聲明一個Binding須要提供一個QueueName，ExchangeName和BindingKey。下面咱們就分析一下不一樣的ExchangeType表現出的不一樣路由規則。

生產者在發送消息時，都須要指定一個RoutingKey和Exchange，Exchange在接到該RoutingKey之後，會判斷該ExchangeType:

a) 若是是Direct類型，則會將消息中的RoutingKey與該Exchange關聯的全部Binding中的BindingKey進行比較，若是相等，則發送到該Binding對應的Queue中。

b) 若是是 Fanout 類型，則會將消息發送給全部與該 Exchange 定義過 Binding 的全部 Queues 中去，實際上是一種廣播行爲。

c)若是是Topic類型，則會按照正則表達式，對RoutingKey與BindingKey進行匹配，若是匹配成功，則發送到對應的Queue中。

3. 消費者訂閱消息

在RabbitMQ中消費者有2種方式獲取隊列中的消息:

a) 一種是經過basic.consume命令，訂閱某一個隊列中的消息,channel會自動在處理完上一條消息以後，接收下一條消息。（同一個channel消息處理是串行的）。除非關閉channel或者取消訂閱，不然客戶端將會一直接收隊列的消息。

b) 另一種方式是經過basic.get命令主動獲取隊列中的消息，可是絕對不能夠經過循環調用basic.get來代替basic.consume，這是由於basic.get RabbitMQ在實際執行的時候，是首先consume某一個隊列，而後檢索第一條消息，而後再取消訂閱。若是是高吞吐率的消費者，最好仍是建議使用basic.consume。

若是有多個消費者同時訂閱同一個隊列的話，RabbitMQ是採用循環的方式分發消息的，每一條消息只能被一個訂閱者接收。例如，有隊列Queue，其中ClientA和ClientB都Consume了該隊列，MessageA到達隊列後，被分派到ClientA，ClientA回覆服務器收到響應，服務器刪除MessageA；再有一條消息MessageB抵達隊列，服務器根據「循環推送」原則，將消息會發給ClientB，而後收到ClientB的確認後，刪除MessageB；等到再下一條消息時，服務器會再將消息發送給ClientA。

這裏咱們能夠看出，消費者再接到消息之後，都須要給服務器發送一條確認命令，這個便可以在handleDelivery裏顯示的調用basic.ack實現，也能夠在Consume某個隊列的時候，設置autoACK屬性爲true實現。這個ACK僅僅是通知服務器能夠安全的刪除該消息，而不是通知生產者，與RPC不一樣。若是消費者在接到消息之後還沒來得及返回ACK就斷開了鏈接，消息服務器會重傳該消息給下一個訂閱者，若是沒有訂閱者就會存儲該消息。

既然RabbitMQ提供了ACK某一個消息的命令，固然也提供了Reject某一個消息的命令。當客戶端發生錯誤，調用basic.reject命令拒絕某一個消息時，能夠設置一個requeue的屬性，若是爲true，則消息服務器會重傳該消息給下一個訂閱者；若是爲false，則會直接刪除該消息。固然，也能夠經過ack，讓消息服務器直接刪除該消息而且不會重傳。

4. 持久化：

Rabbit MQ默認是不持久隊列、Exchange、Binding以及隊列中的消息的，這意味着一旦消息服務器重啓，全部已聲明的隊列，Exchange，Binding以及隊列中的消息都會丟失。經過設置Exchange和MessageQueue的durable屬性爲true，可使得隊列和Exchange持久化，可是這還不能使得隊列中的消息持久化，這須要生產者在發送消息的時候，將delivery mode設置爲2，只有這3個所有設置完成後，才能保證服務器重啓不會對現有的隊列形成影響。這裏須要注意的是，只有durable爲true的Exchange和durable爲ture的Queues才能綁定，不然在綁定時，RabbitMQ都會拋錯的。持久化會對RabbitMQ的性能形成比較大的影響，可能會降低10倍不止。

5. 事務：

對事務的支持是AMQP協議的一個重要特性。假設當生產者將一個持久化消息發送給服務器時，由於consume命令自己沒有任何Response返回，因此即便服務器崩潰，沒有持久化該消息，生產者也沒法獲知該消息已經丟失。若是此時使用事務，即經過txSelect()開啓一個事務，而後發送消息給服務器，而後經過txCommit()提交該事務，便可以保證，若是txCommit()提交了，則該消息必定會持久化，若是txCommit()還未提交即服務器崩潰，則該消息不會服務器就收。固然Rabbit MQ也提供了txRollback()命令用於回滾某一個事務。

6. Confirm機制：

使用事務當然能夠保證只有提交的事務，纔會被服務器執行。可是這樣同時也將客戶端與消息服務器同步起來，這背離了消息隊列解耦的本質。Rabbit MQ提供了一個更加輕量級的機制來保證生產者能夠感知服務器消息是否已被路由到正確的隊列中——Confirm。若是設置channel爲confirm狀態，則經過該channel發送的消息都會被分配一個惟一的ID，而後一旦該消息被正確的路由到匹配的隊列中後，服務器會返回給生產者一個Confirm，該Confirm包含該消息的ID，這樣生產者就會知道該消息已被正確分發。對於持久化消息，只有該消息被持久化後，纔會返回Confirm。Confirm機制的最大優勢在於異步，生產者在發送消息之後，便可繼續執行其餘任務。而服務器返回Confirm後，會觸發生產者的回調函數，生產者在回調函數中處理Confirm信息。若是消息服務器發生異常，致使該消息丟失，會返回給生產者一個nack，表示消息已經丟失，這樣生產者就能夠經過重發消息，保證消息不丟失。Confirm機制在性能上要比事務優越不少。可是Confirm機制，沒法進行回滾，就是一旦服務器崩潰，生產者沒法獲得Confirm信息，生產者其實自己也不知道該消息吃否已經被持久化，只有繼續重發來保證消息不丟失，可是若是原先已經持久化的消息，並不會被回滾，這樣隊列中就會存在兩條相同的消息，系統須要支持去重。

其餘：

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。

Linux下安裝RabbitMQ

標籤： rabbitmq 消息隊列

2015-01-24 23:12 777人閱讀評論(0) 收藏舉報

1. 安裝erlang：

下載所需的源碼：wget http://erlang.org/download/otp_src_R13B04.tar.gz wget http://erlang.org/download/otp_src_R13B04.tar.gz

而後./configure && make && make install

注：在configure以後發現有如下提示信息：

(1) odbc : ODBC library – link check failed

須要yum -y install unixODBC unixODBC-devel

(2)「 wxWidgets not found, wx will NOT be usable」及「fop is missing.」這兩個能夠忽略。

安裝完畢，在命令行鍵入erl，將會出現以下命令行：

Erlang R13B04 (erts-5.7.5) [source] [smp:2:2] [rq:2][async-threads:0] [hipe] [kernel-poll:false]

Eshell V5.7.5

(abort with ^G)
1>

2.安裝RabbitMQ

RabbitMQ的安裝有不少版本，咱們使用Generic Unix版本。

在http://www.rabbitmq.com/install-generic-unix.html中下載http://www.rabbitmq.com/releases/rabbitmq-server/v3.4.3/rabbitmq-server-generic-unix-3.4.3.tar.gz並解壓到Linux中的任意目錄(/mysoftware/rabbitmq_server-3.4.3/)，解壓後便可使用。

3.RabbitMQ開啓與關閉

進入rabbitmq安裝目錄下的sbin目錄，分別執行如下命令：

[plain] view plain copy print ?

開啓RabbitMQ服務：.
./rabbitmq-server -detached (-detached爲可選參數，表示後臺開啓)
開啓RabbitMQ管理工具，經過瀏覽器訪問http://localhost:15672使用：.
./rabbitmq-plugins enable rabbitmq_management （3.3.1版本之後默認不容許用guest遠程管理。能夠手動建立管理員賬號來遠程管理，具體參見4.4）
關閉RabbitMQ服務：
./rabbitmqctl stop

開啓RabbitMQ服務：.
./rabbitmq-server -detached (-detached爲可選參數，表示後臺開啓)

開啓RabbitMQ管理工具，經過瀏覽器訪問http://localhost:15672使用：.
./rabbitmq-plugins enable rabbitmq_management （3.3.1版本之後默認不容許用guest遠程管理。能夠手動建立管理員賬號來遠程管理，具體參見4.4）

關閉RabbitMQ服務：
./rabbitmqctl stop

注意：若是是登陸遠程管理界面(好比物理機訪問虛擬機的管理界面時)，除了關閉防火牆開通相應容許的端口之外，也不能用guest/guest訪問，而應該新建立用戶並賦予管理員權限纔可訪問

增長用戶：rabbitmqctl add_user ddd ddd

設置管理員權限： rabbitmqctl set_user_tags diego administrator 這樣才能夠訪問web頁面提供的管理頁面

./rabbitmqctl set_permissions -p / diego '.*' '.*' '.*' 給用戶diego加上有訪問虛擬空間/的權限

4.設置RabbitMQ管理權限

Rabbitmq服務器的主要經過rabbitmqctl和rabbimq-plugins兩個工具來管理，如下是一些經常使用功能。

1）. 服務器啓動與關閉

啓動: rabbitmq-server –detached

關閉:rabbitmqctl stop

若單機有多個實例，則在rabbitmqctlh後加–n 指定名稱

2）. 插件管理

開啓某個插件：rabbitmq-pluginsenable xxx

關閉某個插件：rabbitmq-pluginsdisablexxx

注意：重啓服務器後生效。

3）.virtual_host管理

新建virtual_host: rabbitmqctladd_vhost xxx

撤銷virtual_host:rabbitmqctl delete_vhost xxx

4）. 用戶管理

新建用戶：rabbitmqctl add_user xxx pwd

刪除用戶: rabbitmqctl delete_user xxx

改密碼: rabbimqctl change_password {username} {newpassword}

設置用戶角色：rabbitmqctl set_user_tags {username} {tag ...}

Tag能夠爲 administrator,monitoring, management

5）. 權限管理

權限設置：set_permissions [-pvhostpath] {user} {conf} {write} {read}

Vhostpath

Vhost路徑

user

用戶名

Conf

一個正則表達式match哪些配置資源可以被該用戶訪問。

Write

一個正則表達式match哪些配置資源可以被該用戶讀。

Read

一個正則表達式match哪些配置資源可以被該用戶訪問。

6）. 獲取服務器狀態信息

服務器狀態：rabbitmqctl status

隊列信息：rabbitmqctl list_queues[-p vhostpath] [queueinfoitem ...]

Queueinfoitem能夠爲：name，durable，auto_delete，arguments，messages_ready，

messages_unacknowledged，messages，consumers，memory

Exchange信息：rabbitmqctllist_exchanges[-p vhostpath] [exchangeinfoitem ...]

Exchangeinfoitem有：name，type，durable，auto_delete，internal，arguments.

Binding信息：rabbitmqctllist_bindings[-p vhostpath] [bindinginfoitem ...]

Bindinginfoitem有：source_name，source_kind，destination_name，destination_kind，routing_key，arguments

Connection信息：rabbitmqctllist_connections [connectioninfoitem ...]

Connectioninfoitem有：recv_oct，recv_cnt，send_oct，send_cnt，send_pend等。

Channel信息：rabbitmqctl list_channels[channelinfoitem ...]

Channelinfoitem有consumer_count，messages_unacknowledged，messages_uncommitted，acks_uncommitted，messages_unconfirmed，prefetch_count，client_flow_blocked

rabbitmq 實現原理

AMQP(高級消息隊列協議 Advanced Message Queue Protocol)

AMQP當中有四個概念很是重要: 虛擬主機（virtual host），交換機（exchange），隊列（queue）和綁定（binding）。一個虛擬主機持有一組交換機、隊列和綁定。爲何須要多個虛擬主機呢？很簡單，RabbitMQ當中，用戶只能在虛擬主機的粒度進行權限控制。所以，若是須要禁止A組訪問B組的交換機/隊列/綁定，必須爲A和B分別創建一個虛擬主機。每個RabbitMQ服務器都有一個默認的虛擬主機「/」。

Producer 要產生消息必需要建立一個 Exchange ，Exchange 用於轉發消息，可是它不會作存儲，若是沒有 Queue bind 到 Exchange 的話，它會直接丟棄掉 Producer 發送過來的消息，固然若是消息老是發送過去就被直接丟棄那就沒有什麼意思了，一個 Consumer 想要接受消息的話，就要建立一個 Queue ，並把這個 Queue bind 到指定的 Exchange 上，而後 Exchange 會把消息轉發到 Queue 那裏，Queue 會負責存儲消息，Consumer 能夠經過主動 Pop 或者是 Subscribe 以後被動回調的方式來從 Queue 中取得消息。

Exchange，Queue，RoutingKey

藍色-- Client（相對於Rabbitmq Server來講）

綠色--Exchange

紅色—Queue

- 交換器（Exchange），它是發送消息的實體。

- 隊列（Queue），這是接收消息的實體。

- 綁定器（Bind），將交換器和隊列鏈接起來，而且封裝消息的路由信息。

Exchange指向Queue的黑色線—RoutingKey，能夠將它簡單的理解爲一條鏈接Exchange和Queue的路線

Exchange和Queue都須要經過channel來進行定義，而RoutingKey則只須要在binding時取個名字就好了。

這一塊的理解是不正確的，

Exchange Queue RoutingKey關係說明：

Exchange Name	Queue Name	Routing Key
	test.queue	test.queue
	test.queue2	test.queue2
test.exchange	test.queue	test.routingkey
test.exchange	test.queue2	test.routingkey
test.exchange	test.queue	test.routingkey2
test.exchange1	test.queue	test.routingkey

由結果能夠看出，由Exchange，Queue，RoutingKey三個才能決定一個從Exchange到Queue的惟一的線路。

左邊的Client向右邊的Client發送消息，流程：

1，獲取Conection

2，獲取Channel

3，定義Exchange，Queue

4，使用一個RoutingKey將Queue Binding到一個Exchange上

5，經過指定一個Exchange和一個RoutingKey來將消息發送到對應的Queue上，

6，接收方在接收時也是獲取connection，接着獲取channel，而後指定一個Queue直接到它關心的Queue上取消息，它對Exchange，RoutingKey及如何binding都不關心，到對應的Queue上去取消息就OK了(這個是重點，接收方，只關心queue的名字便可，其餘的都不關心！！！！！)

換句話說：

發送放關心：ExchangName， QueueName，BindingKey

接收方關心：QuueuName

一個Client發送消息，哪些Client能夠收到消息，其核心就在於Exchange，RoutingKey，Queue的關係上。