分佈式系統消息中間件——RabbitMQ的使用進階篇

一 mandatory 參數

上一篇文章中咱們知道，生產者將消息發送到RabbitMQ的交換器中經過RoutingKey與BindingKey的匹配將之路由到具體的隊列中以供消費者消費。那麼當咱們經過匹配規則找不到隊列的時候，消息將何去何從呢?Rabbit給咱們提供了兩種方式。mandatory與備份交換器。

mandatory參數是channel.BasicPublish方法中的參數。其主要功能是消息傳遞過程當中不可達目的地時將消息返回給生產者。當mandatory 參數設爲true 時，交換器沒法根據自身的類型和路由鍵找到一個符合條件的隊列，那麼RabbitMQ 會調用BasicReturn 命令將消息返回給生產者。當mandatory 參數設置爲false 時。則消息直接被丟棄。其運轉流程與實現代碼以下(以C# RabbitMQ.Client 3.6.9爲例):
複製代碼

 

 

 

二 備份交換器

 

當消息不能路由到隊列時，經過mandatory設置參數,咱們能夠將消息返回給生產者處理。但這樣會有一個問題，就是生產者須要開一個回調的函數來處理不能路由到的消息，這無疑會增長生產者的處理邏輯。備份交換器(Altemate Exchange)則提供了另外一種方式來處理不能路由的消息。備份交換器能夠將未被路由的消息存儲在RabbitMQ中，在須要的時候去處理這些消息。其主要實現代碼以下:
複製代碼

 

 

 

 

備份交換器其實和普通的交換器沒有太大的區別，爲了方便使用，建議設置爲fanout類型，若設置爲direct 或者topic的類型。須要注意的是，消息被從新發送到備份交換器時的路由鍵和從生產者發出的路由鍵是同樣的。考慮這樣一種狀況，若是備份交換器的類型是direct,而且有一個與其綁定的隊列，假設綁定的路由鍵是key1，當某條攜帶路由鍵爲key2 的消息被轉發到這個備份交換器的時候，備份交換器沒有匹配到合適的隊列，則消息丟失。若是消息攜帶的路由鍵爲keyl，則能夠存儲到隊列中。 對於備份交換器，有如下幾種特殊狀況:

1.若是設置的備份交換器不存在，客戶端和RabbitMQ 服務端都不會有異常出現，此時消息會丟失。

2.若是備份交換器沒有綁定任何隊列，客戶端和RabbitMQ 服務端都不會有異常出現，此時消息會丟失。

3.若是備份交換器沒有任何匹配的隊列，客戶端和RabbitMQ 服務端都不會有異常出現，此時消息會丟失。

4.若是備份交換器和mandatory參數一塊兒使用，那麼mandatory參數無效。

三 過時時間(TTL)

3.1 設置消息的TTL 目前有兩種方法能夠設置消息的TTL。第一種方法是經過隊列屬性設置，隊列中全部消息都有相同的過時時間。第二種方法是對消息自己進行單獨設置，每條消息的TTL能夠不一樣。若是兩種方法一塊兒使用，則消息的TTL 以二者之間較小的那個數值爲準。消息在隊列中的生存時間一旦超過設置的TTL值時，就會變成"死信" (Dead Message) ，消費者將沒法再收到該消息。(有關死信隊列請往下看)

經過隊列屬性設置消息TTL的方法是在channel.QueueDeclare方法中加入x-message-ttl參數實現的，這個參數的單位是毫秒。示例代碼下:
複製代碼

 

若是不設置TTL.則表示此消息不會過時;若是將TTL設置爲0 ，則表示除非此時能夠直接將消息投遞到消費者，不然該消息會被當即丟棄(或由死信隊列來處理)。

 

針對每條消息設置TTL的方法是在channel.BasicPublish方法中加入Expiration的屬性參數，單位爲毫秒。關鍵代碼以下：
複製代碼

 

注意:對於第一種設置隊列TTL屬性的方法，一旦消息過時，就會從隊列中抹去，而在第二種方法中，即便消息過時，也不會立刻從隊列中抹去，由於每條消息是否過時是在即將投遞到消費者以前斷定的。Why?在第一種方法裏，隊列中己過時的消息確定在隊列頭部， RabbitMQ 只要按期從隊頭開始掃描是否有過時的消息便可。而第二種方法裏，每條消息的過時時間不一樣，若是要刪除全部過時消息勢必要掃描整個隊列，因此不如等到此消息即將被消費時再斷定是否過時，若是過時再進行刪除便可。

 

3.2 設置隊列的TTL 注意，這裏和上述經過隊列設置消息的TTL不一樣。上面刪除的是消息，而這裏刪除的是隊列。經過channel.QueueDeclare 方法中的x-expires參數能夠控制隊列被自動刪除前處於未使用狀態的時間。這個未使用的意思是隊列上沒有任何的消費者，隊列也沒有被從新聲明，而且在過時時間段內也未調用過channel.BasicGet命令。

設置隊列裏的TTL能夠應用於相似RPC方式的回覆隊列，在RPC中，許多隊列會被建立出來，可是倒是未被使用的(有關RabbitMQ實現RPC請往下看)。RabbitMQ會確保在過時時間到達後將隊列刪除，可是不保障刪除的動做有多及時。在RabbitMQ 重啓後， 持久化的隊列的過時時間會被從新計算。用於表示過時時間的x-expires參數以毫秒爲單位， 井且服從和x-message-ttl同樣的約束條件，不一樣的是它不能設置爲0(會報錯)。
複製代碼

示例代碼以下：

 

四 死信隊列 DLX(Dead-Letter-Exchange)死信交換器，當消息在一個隊列中變成死信以後，它能被從新被髮送到另外一個交換器中，這個交換器就是DLX ，綁定DLX的隊列就稱之爲死信隊列。 消息變成死信主要有如下幾種狀況:

 

在此我向你們推薦一個架構學習交流羣。交流學習羣號：874811168 裏面會分享一些資深架構師錄製的視頻錄像：有Spring，MyBatis，Netty源碼分析，高併發、高性能、分佈式、微服務架構的原理，JVM性能優化、分佈式架構等這些成爲架構師必備的知識體系。還能領取免費的學習資源，目前受益良多

消息被拒絕(BasicReject/BasicNack) ，井且設置requeue 參數爲false;(消費者確認機制將會在下一篇文章中涉及) 消息過時; 隊列達到最大長度。 DLX也是一個正常的交換器，和通常的交換器沒有區別，它能在任何的隊列上被指定，實際上就是設置某個隊列的屬性。當這個隊列中存在死信時，RabbitMQ 就會自動地將這個消息從新發布到設置的DLX上去，進而被路由到另外一個隊列，即死信隊列。能夠監聽這個隊列中的消息、以進行相應的處理。

經過在channel.QueueDeclare 方法中設置x-dead-letter-exchange參數來爲這個隊列添加DLX。其示例代碼以下:
複製代碼

 

如下爲死信隊列的運轉流程:

 

 

 

五 延遲隊列 RabbitMQ自己並未提供延遲隊列的功能。延遲隊列是一個邏輯上的概念，能夠經過過時時間+死信隊列來模擬它的實現。延遲隊列的邏輯架構大體以下:

 

 

生產者將消息發送到過時時間爲n的隊列中，這個隊列並未有消費者來消費消息，當過時時間到達時，消息會經過死信交換器被轉發到死信隊列中。而消費者從死信隊列中消費消息。這個時候就達到了生產者發佈了消息在講過了n時間後消費者消費了消息，起到了延遲消費的做用。

 延遲隊列在咱們的項目中能夠應用於不少場景，如：下單後兩個消息取消訂單，七天自動收貨，七天自動好評，密碼凍結後24小時解凍，以及在分佈式系統中消息補償機制(1s後補償,10s後補償，5m後補償......)。
複製代碼

 

 

六 優先級隊列 就像咱們生活中的「特殊」人士同樣，咱們的業務上也存在一些「特殊」消息，可能須要優先進行處理，在生活上咱們可能會對這部分特殊人士開闢一套VIP通道，而Rabbit一樣也有這樣的VIP通道(前提是在3.5的版本之後)，即優先級隊列，隊列中的消息會有優先級優先級高的消息具有優先被消費的特權。針對這些VIP消息，咱們只需作兩件事:

咱們只需作兩件事情：

將隊列聲明爲優先級隊列，即在建立隊列的時候添加參數 x-max-priority 以指定最大的優先級，值爲0-255（整數）。 爲優先級消息添加優先級。 其示例代碼以下:

 

注意：沒有指定優先級的消息會將優先級以0對待。 對於超過優先級隊列所定最大優先級的消息，優先級以最大優先級對待。對於相同優先級的消息，後進的排在前面。若是在消費者的消費速度大於生產者的速度且Broker 中沒有消息堆積的狀況下， 對發送的消息設置優先級也就沒有什麼實際意義。由於生產者剛發送完一條消息就被消費者消費了，那麼就至關於Broker 中至多隻有一條消息，對於單條消息來講優先級是沒有什麼意義的。

 

關於優先級隊列，好像違背了隊列這種數據結構先進先出的原則，其具體是怎麼實現的在這裏就不過多討論。有興趣的能夠本身研究研究。後續可能也會有相關的文章來分析其原理。
複製代碼

七 RPC 實現 RPC,是Remote Procedure Call 的簡稱，即遠程過程調用。它是一種經過網絡從遠程計算機上請求服務，而不須要了解底層網絡的技術。RPC 的主要功用是讓構建分佈式計算更容易，在提供強大的遠程調用能力時不損失本地調用的語義簡潔性。

有關RPC很少介紹，這裏咱們主要介紹RabbitMQ如何實現RPC。RabbitMQ 能夠實現很簡單的RPC。客戶端發送請求消息，服務端回覆響應的消息，爲了接收響應的消息，咱們須要在請求消息中發送一個回調隊列(可使用默認的隊列)。其服務器端實現代碼以下:
複製代碼

 

 

 

客戶端實現代碼以下:

 

 

以上是Rabbit客戶端本身幫咱們封裝好的Rpc客戶端與服務端的邏輯。固然咱們也能夠本身實現，主要是藉助於BasicProperties的兩個參數。

 

ReplyTo: 一般用來設置一個回調隊列。 CorrelationId : 用來關聯請求(request) 和其調用RPC 以後的回覆(response) 。 其處理流程以下:

 

 

當客戶端啓動時，建立一個匿名的回調隊列。 客戶端爲RPC 請求設置2個屬性: ReplyTo用來告知RPC 服務端回覆請求時的目的隊列，即回調隊列; Correlationld 用來標記一個請求。 請求被髮送到RpcQueue隊列中。 RPC 服務端監聽RpcQueue隊列中的請求，當請求到來時，服務端會處理而且把帶有結果的消息發送給客戶端。接收的隊列就是ReplyTo設定的回調隊列。 客戶端監昕回調隊列，當有消息時，檢查Correlationld 屬性，若是與請求匹配，那就是結果了。 結束語 本篇文章簡單介紹了RabbitMQ在咱們項目開發中經常使用的幾種特性。這些特性能夠幫助咱們更好的將Rabbit用於咱們不一樣的業務場景中。這些特性與示例，能夠本身在程序中運行一下，而後經過查看Rabbit提供的web管理界面來驗證其正確性(關於web管理界面很少介紹，相信你們稍微研究研究就能明白)。固然，關於Rabbit的使用，仍有許多地方在本文中沒有說起，如：RabbitMQ的特點——確認機制、持久化......將在下一篇文章中再詳細介紹。

出  處：www.cnblogs.com/hunternet