RabbitMQ基本知識介紹參考范本

1、精品RabbitMQ基礎(chǔ)概念詳細介紹引言你是否遇到過兩個（多個） 系統(tǒng)間需要通過定時任務(wù)來同步某些數(shù)據(jù)？你是否在為異構(gòu)系統(tǒng)的不同進程間相互調(diào)用、通訊的問題而苦惱、掙扎？如果是，那么恭喜你，消息服務(wù)讓你可以很輕松地解決這些問題。消息服務(wù)擅長于解決多系統(tǒng)、異構(gòu)系統(tǒng)間的數(shù)據(jù)交換（消息通知/ 通訊）問題，你也可以把它用于系統(tǒng)間服務(wù)的相互調(diào)用（RPC ）。本文將要介紹的RabbitMQ就是當(dāng)前最主流的消息中間件之一。RabbitMQ簡介AMQP ， 即 Advanced Message Queuing Protocol，高級消息隊列協(xié)議，是應(yīng)用層協(xié)議的一 個開放標準， 為面向消息的中間件設(shè)計。消息中間件

2、主要用于組件之間的解耦，消息的發(fā)送者無需知道消息使用者的存在，反之亦然。AMQP的主要特征是面向消息、隊列、路由（包括點對點和發(fā)布/ 訂閱）、可靠性、安全。RabbitMQ是一個開源的AMQP實現(xiàn)，服務(wù)器端用Erlang語言編寫，支持多種客戶端，如： Python、Ruby 、.NET 、Java 、JMS 、C、PHP 、ActionScript、 XMPP 、STOMP等，支持 AJAX 。用于在分布式系統(tǒng)中存儲轉(zhuǎn)發(fā)消息，在易用性、擴展性、高可用性等方面表現(xiàn)不俗。下面將重點介紹RabbitMQ中的一些基礎(chǔ)概念， 了解了這些概念， 是使用好RabbitMQ的基礎(chǔ)。ConnectionFact

3、ory、Connection、Channel感謝下載載ConnectionFactory、Connection、Channel都是 RabbitMQ對外提供的API 中最基本的對象。 Connection是 RabbitMQ的 socket鏈接，它封裝了socket協(xié)議相關(guān)部分邏輯。ConnectionFactory為 Connection的制造工廠。Channel是我們與RabbitMQ打交道的最重要的一個接口，我們大部分的業(yè)務(wù)操作是在Channel這個接口中完成的，包括定義Queue 、定義 Exchange、綁定Queue與 Exchange、發(fā)布消息等。QueueQueue （隊列）是

4、RabbitMQ的內(nèi)部對象，用于存儲消息，用下圖表示。RabbitMQ中的消息都只能存儲在Queue中，生產(chǎn)者（下圖中的P）生產(chǎn)消息并最終投遞到Queue中，消費者（下圖中的C）可以從Queue中獲取消息并消費。多個消費者可以訂閱同一個Queue ， 這時Queue中的消息會被平均分攤 給多個消費者進行處理，而不是每個消費者都收到所有的消息并處理。Message acknowledgment在實際應(yīng)用中，可能會發(fā)生消費者收到Queue中的消息，但沒有處理完成就宕機（或出現(xiàn)其他 意外）的情況，這種情況下就可能會導(dǎo)致消息丟失。為了避免這種情況發(fā)生，我們可以要求消費者在消費完消息后發(fā)送一個回執(zhí)給Ra

5、bbitMQ， RabbitMQ收到消息回執(zhí)（Message acknowledgment）后才將該消息從Queue中移除；如果RabbitMQ沒有收到回執(zhí)并檢測到消費者的RabbitMQ連接斷開， 則 RabbitMQ會將該消息發(fā)送給其他消費者（如果存在多個消費者）進行處理。這里不存在timeout概念，一個消費者處理消息時間再長也不會導(dǎo)致該消息被發(fā)送給其他消費者，除非它的RabbitMQ連接斷開。這里會產(chǎn)生另外一個問題，如果我們的開發(fā)人員在處理完業(yè)務(wù)邏輯后，忘記發(fā)送回執(zhí)給RabbitMQ，這將會導(dǎo)致嚴重的bug Queue中堆積的消息會越來越多；消費者重啟后會重復(fù)消費這些消息并重復(fù)執(zhí)行業(yè)務(wù)

6、邏輯Message durability如果我們希望即使在RabbitMQ服務(wù)重啟的情況下，也不會丟失消息，我們可以將Queue與Message都設(shè)置為可持久化的（durable），這樣可以保證絕大部分情況下我們的RabbitMQ 消息不會丟失。但依然解決不了小概率丟失事件的發(fā)生（比如RabbitMQ服務(wù)器已經(jīng)接收到生產(chǎn)者的消息，但還沒來得及持久化該消息時RabbitMQ服務(wù)器就斷電了），如果我們需要對這種小概率事件也要管理起來，那么我們要用到事務(wù)。由于這里僅為RabbitMQ的簡單介紹，所以這里將不講解RabbitMQ相關(guān)的事務(wù)。Prefetch count前面我們講到如果有多個消費者同時訂

7、閱同一個Queue中的消息，Queue中的消息會被平攤 給多個消費者。這時如果每個消息的處理時間不同，就有可能會導(dǎo)致某些消費者一直在忙，而另外一些消費者很快就處理完手頭工作并一直空閑的情況。我們可以通過設(shè)置prefetchCount 來限制 Queue 每次發(fā)送給每個消費者的消息數(shù)， 比如我們設(shè)置 prefetchCount=1 ，則 Queue 每次給每個消費者發(fā)送一條消息； 消費者處理完這條消息后 Queue 會再給該消費者發(fā)送一條消息。Exchange在上一節(jié)我們看到生產(chǎn)者將消息投遞到Queue中，實際上這在RabbitMQ中這種事情永遠都不會發(fā)生。 實際的情況是， 生產(chǎn)者將消息發(fā)送到E

8、xchange （交換器， 下圖中的X）， 由 Exchange 將消息路由到一個或多個Queue中（或者丟棄）。Exchange是按照什么邏輯將消息路由到Queue的？這個將在Binding一 節(jié) 介 紹 。 RabbitMQ中的 Exchange有四種類型，不同的類型有著不同的路由策略，這將在Exchange Types一節(jié)介紹。routing key生產(chǎn)者在將消息發(fā)送給 Exchange 的時候，一般會指定一個 routing key ，來指定這個消息的路由規(guī)則，而這個 routing key 需要與 Exchange Type 及 binding key 聯(lián)合使用才能最終生效。在 Ex

9、change Type 與 binding key 固定的情況下 （在正常使用時一般這些內(nèi)容都是固定配置好的），我們的生產(chǎn)者就可以在發(fā)送消息給Exchange時， 通過指定routing key來決定消息流向哪里。RabbitMQ為 routing key設(shè)定的長度限制為255 bytes。BindingRabbitMQ中通過Binding將 Exchange與 Queue關(guān)聯(lián)起來，這樣RabbitMQ就知道如何正確地將消息路由到指定的Queue了。Binding key在綁定（ Binding）Exchange與 Queue的同時，一般會指定一個binding key；消費者將消息發(fā)送給 E

10、xchange時，一般會指定一個routing key；當(dāng) binding key與 routing key相匹配時， 消息將會被路由到對應(yīng)的Queue中。這個將在Exchange Types章節(jié)會列舉實際的例子加以說 明。在綁定多個Queue到同一個Exchange的時候，這些Binding允許使用相同的binding key。binding key并不是在所有情況下都生效，它依賴于Exchange Type，比如 fanout類型的Exchange就會無視binding key，而是將消息路由到所有綁定到該Exchange的 Queue 。Exchange TypesRabbitMQ常用的

11、Exchange Type有 fanout、direct 、topic 、headers這四種（ AMQP規(guī)范里還提到兩種Exchange Type，分別為system與自定義，這里不予以描述），下面分別進行介紹。fanoutfanout類型的Exchange路由規(guī)則非常簡單，它會把所有發(fā)送到該Exchange的消息路由到所有與它綁定的Queue中。上圖中，生產(chǎn)者（P）發(fā)送到Exchange（ X）的所有消息都會路由到圖中的兩個Queue ，并最終被兩個消費者（C1 與 C2 ）消費。directdirect類型的 Exchange路由規(guī)則也很簡單，它會把消息路由到那些binding key與

12、 routing key完全匹配的Queue中。以上圖的配置為例，我們以routingKey=”error ”發(fā)送消息到Exchange ，則消息會路由到Queue1（ amqp.gen-S9b，這是由RabbitMQ自動生成的Queue名稱）和Queue2（amqp.gen-Agl）；如果我們以routingKey=”info ”或 routingKey=”warning”來發(fā)送消息，則消息只會路由到Queue2。如果我們以其他routingKey發(fā)送消息，則消息不會路由到這兩個Queue中。topic前面講到 direct 類型的 Exchange 路由規(guī)則是完全匹配 binding ke

13、y 與 routing key ，但這種嚴格的匹配方式在很多情況下不能滿足實際業(yè)務(wù)需求。topic 類型的 Exchange 在匹配規(guī)則上進行了擴展，它與 direct 類型的 Exchage 相似，也是將消息路由到 binding key 與 routing key 相匹配的 Queue 中，但這里的匹配規(guī)則有些不同，它約定：?routing key為一個句點號“. ”分隔的字符串 （我們將被句點號“. ”分隔開的每一段獨立的字符串稱為一個單詞），如“stock.usd.nyse”、 “nyse.vmw”、“quick.orange.rabbit”?binding key與 routing

14、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 一次，雖然這個 routi

15、ngKey與 Q2 的兩個 bindingKey都匹配）； routingKey=”quick.brown.fox”、 routingKey=”orange”、routingKey=”quick.orange.male.rabbit”的消息將會被丟棄，因為 它們沒有匹配任何bindingKey。headersheaders類型的 Exchange不依賴于routing key與 binding key的匹配規(guī)則來路由消息，而是根據(jù)發(fā)送的消息內(nèi)容中的headers屬性進行匹配。在綁定 Queue與 Exchange時指定一組鍵值對；當(dāng)消息發(fā)送到Exchange時，RabbitMQ會取到該消息的h

16、eaders（也是一個鍵值對的形式），對比其中的鍵值對是否完全匹配Queue與Exchange綁定時指定的鍵值對；如果完全匹配則消息會路由到該Queue ，否則不會路由到該Queue 。該類型的Exchange沒有用到過（不過也應(yīng)該很有用武之地），所以不做介紹。RPCMQ本身是基于異步的消息處理，前面的示例中所有的生產(chǎn)者（P）將消息發(fā)送到RabbitMQ后不會知道消費者（C）處理成功或者失敗（甚至連有沒有消費者來處理這條消息都不知道）。 但實際的應(yīng)用場景中，我們很可能需要一些同步處理，需要同步等待服務(wù)端將我的消息處理完成后再進行下一步處理。這相當(dāng)于RPC（ Remote Procedure Call，遠程過程調(diào)用） 。在 RabbitMQ中也支持RPC 。RabbitMQ中實現(xiàn)RPC 的機制是：?客戶端發(fā)送請求（消息）時，在消息的屬性（MessageProperties，在 AMQP協(xié)議中定義 了 14 中 properties，這些屬性會隨著消息一起發(fā)送）中設(shè)置兩個值replyTo （一個 Queue 名稱，用于告訴服務(wù)器處理完成后將通知我的消息發(fā)送到這個Queue中）和 correlationId（此次請求的標識號，服務(wù)器處理完成后