個人學習SpringBoot系列 RabbitMQ篇

Github Link: https://github.com/panjianlong13/SpringBoot-SpringCloud/tree/master/spring-boot-rabbitmq

RabbitMQ介紹

RabbitMQ 是什麼

MQ 是什麼？佇列是什麼，MQ 我們可以理解為訊息佇列，佇列我們可以理解為管道。

場景：

    1.其實我們在雙11的時候，當我們凌晨大量的秒殺和搶購商品，然後去結算的時候，就會發現，介面會提醒我們，讓我們稍等，以及一些友好的圖片文字提醒。而不是像前幾年的時代，動不動就頁面卡死，報錯等來呈現給使用者。在這業務場景中，我們就可以採用佇列的機制來處理，因為同時結算就只能達到這麼多。

    2.在我們平時的超市中購物也是一樣，當我們在結算的時候，並不會一窩蜂一樣湧入收銀臺，而是排隊結算。這也是佇列機制。排隊一個接著一個的處理，不能插隊。

RabbitMQ是一個由erlang開發的AMQP（Advanced Message Queue ）的開源實現，支援多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP.

AMQP，即Advanced message Queuing Protocol，高階訊息佇列協議，是應用層協議的一個開放標準，為面向訊息的中介軟體設計。訊息中介軟體主要用於元件之間的解耦，訊息的傳送者無需知道訊息使用者的存在。AMQP

Queue

Queue（佇列）是RabbitMQ的內部物件，用於儲存訊息，用下圖表示。

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

多個消費者可以訂閱同一個Queue，這時Queue中的訊息會被平均分攤給多個消費者進行處理，而不是每個消費者都收到所有的訊息並處理。 

Message acknowledgment

在實際應用中，可能會發生消費者收到Queue中的訊息，但沒有處理完成就宕機（或出現其他意外）的情況，這種情況下就可能會導致訊息丟失。為了避免這種情況發生，我們可以要求消費者在消費完訊息後傳送一個回執給RabbitMQ，RabbitMQ收到訊息回執（Message acknowledgment）後才將該訊息從Queue中移除；如果RabbitMQ沒有收到回執並檢測到消費者的RabbitMQ連線斷開，則RabbitMQ會將該訊息傳送給其他消費者（如果存在多個消費者）進行處理。這裡不存在timeout概念，一個消費者處理訊息時間再長也不會導致該訊息被髮送給其他消費者，除非它的RabbitMQ連線斷開。 這裡會產生另外一個問題，如果我們的開發人員在處理完業務邏輯後，忘記傳送回執給RabbitMQ，這將會導致嚴重的bug——Queue中堆積的訊息會越來越多；消費者重啟後會重複消費這些訊息並重復執行業務邏輯…

另外pub message是沒有ack的。

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中，但這裡的匹配規則有些不同，它約定：

1.routing key為一個句點號“. ”分隔的字串（我們將被句點號“. ”分隔開的每一段獨立的字串稱為一個單詞），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”

2.binding key與routing key一樣也是句點號“. ”分隔的字串

3.binding key中可以存在兩種特殊字元“*”與“#”，用於做模糊匹配，其中“*”用於匹配一個單詞，“#”用於匹配多個單詞（可以是零個）

以上圖中的配置為例，routingKey=”quick.orange.rabbit”的訊息會同時路由到Q1與Q2，routingKey=”lazy.orange.fox”的訊息會路由到Q1與Q2，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 的機制是：

1.客戶端傳送請求（訊息）時，在訊息的屬性（MessageProperties ，在AMQP 協議中定義了14中properties ，這些屬性會隨著訊息一起傳送）中設定兩個值replyTo （一個Queue 名稱，用於告訴伺服器處理完成後將通知我的訊息傳送到這個Queue 中）和correlationId （此次請求的標識號，伺服器處理完成後需要將此屬性返還，客戶端將根據這個id瞭解哪條請求被成功執行了或執行失敗）

2.伺服器端收到訊息並處理

3.伺服器端處理完訊息後，將生成一條應答訊息到replyTo 指定的Queue ，同時帶上correlationId 屬性

4.客戶端之前已訂閱replyTo 指定的Queue ，從中收到伺服器的應答訊息後，根據其中的correlationId 屬性分析哪條請求被執行了，根據執行結果進行後續業務處理

RabbitMQ 單主實戰(Mac)

Rabbit安裝

brew install rabbitmq

## 進入安裝目錄
cd /usr/local/Cellar/rabbitmq/3.7.9

## 啟動
brew services start rabbitmq

## 當前視窗啟動
rabbitmq-server

進入控制檯: http://localhost:15672/

使用者名稱和密碼：guest,guest

新增Admin賬號

## 新增賬號
./rabbitmqctl add_user admin admin
## 新增訪問許可權
./rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"
## 設定超級許可權
./rabbitmqctl set_user_tags admin administrator

使用Admin登入

新建SpringBoot專案

在pom.xml中新增依賴

配置檔案application.properties

啟動專案後分別測試不同type(Direct,Fanout,Topic)