消息隊列那么多，為什么建議深入了解下RabbitMQ？

本文轉載自查看原文 2021-10-18 23:30 19158 消息隊列/ 分布式

你為啥要在項目中選擇xxx消息中間件？

提起消息隊列，也許你的腦海里會不自覺地蹦出好多概念：JMS、Kafka、RocketMQ、AMQP、RabbitMQ、ActiveMQ、Pulsar、Redis Stream...如果你的項目中恰好用到了其中的一個消息中間件，那么你出去面試或者與同事交流技術的時候，對方很大概率會問你：為啥要選擇xxx消息中間件？

如果你剛好只了解你正在用的消息中間件，那么你只能回答：因為只會xxx...

不...這絕對不是你想要的結局！在對方准備看你的笑話之前，你也許可以主動發起還擊，把所有的框架的誕生背景、優缺點，適用場景等都說一遍，從概念到原理，從特性到源碼。在說完了之后，為了不讓對方感覺到尷尬，你應該故意停頓片刻，暗示對方自己不想再聊這個話題了，好讓對方有喘息的機會，讓他趁早切換話題，給他一個台階下。

為了讓自己能有如此實力，你務必要對這些常見的消息中間件有比較深入的了解。

我們先來看看這些技術的發展史。

MQ技術發展史：

如下圖所示：

操作系統中的消息隊列：在操作系統里面，我們可以通過消息隊列實現兩個或多個進程/線程之間的異步通信，發送方和接收方不需要同時與消息隊列交互。放置在隊列中的消息會一直存儲，直到接收者取回它們。消息隊列對可以在單個消息中傳輸的數據大小和隊列能夠存儲的消息數量上限有隱式或顯式限制；
TIB：1983年，一位來自孟買的工程師Vivek Ranadive提出了一個問題：為什么沒有通用的軟件總線--一種通信系統，可以將信息從一個應用程序傳遞到另一個應用程序呢？最終，Vivek Ranadive創辦了Teknekron公司，在1985年實現了第一個消息中間件：Teknekron的The Information Bus(TIB)；
MQSeries：TIB受到了企業的歡迎，同時這也被IBM看在眼里，於是他們決定研發自己的消息隊列軟件。最終，在1993年，IBM推出了面向消息中間件的產品，MQSeries，2002年更名為WebSphereMQ，2014年更名為IBM MQ；
MSMQ：這么好的發財機會怎么能錯過呢，於是微軟也加入了競爭隊伍，並在1997年發布了自家的消息中間件產品：MSMQ；
JMS：這些巨頭推出的消息中間件價格昂貴，一般只應用於大型組織機構。並且，由於商業壁壘，MQ廠商們只關注於應用互通的問題，而不會去考慮創建標准來實現不同的MQ產品之間的互通。為了打破這個壁壘，於是JMS誕生了;
- JMS，即Java消息服務(Java Message Service)，是由Sun公司早期提出的消息標准，為Java提供統一的消息操作，是Java平台中關於面向消息中間件的接口；
- JMS是一種與廠商無關的API，類似於JDBC(Java Database Connectivity)，用來訪問消息系統和收發消息的編程API；
- 不過JMS畢竟是在真實的消息中間件API上面做了一層適配，各個消息中間件的實現仍舊是沒有一個規范，最終會暴露出問題，使得程序更加混亂與脆弱。此刻，我們需要的是一種消息通信標准；
AMQP：在2004至2006年，摩根大通在着手設計AMQP，最終，與其他公司(Cisco, Red Hat, iMatix等)成立了AMQP工作組，越來越多的公司參與進來，最終在2006年制定了AMQP的公開標准，由此，AMQP登上了歷史的舞台，大家可以基於此標准來實現消息中間件，不受任何開發語言、產品等的條件限制；
RabbitMQ：RabbitMQ最初就是一個實現了AMQP的消息中間件，本文我們會詳細介紹這個家伙；
Kafka： Kafka是一種分布式流式系統，被設計為能夠作為一個統一平台來處理大型公司可能擁有的所有實時數據饋送。為此，它必須具有高吞吐量才能支持大容量事件流，例如實時日志聚合；
- RabbitMQ是基於隊列和交換器的消息中間件，而Kafka是使用分區事務日志來實現存儲層的分布式流式系統；
- Kafka不存在隊列，而是按照主題存儲記錄集，並且為每個主題維護一個消息分區日志；
- Kafka中消費者自己維護消息的消費偏移量，支持持久訂閱和臨時訂閱(重啟后丟失偏移)；
- Kafka中的消息是按照預設的時間進行持久化的，而不是根據消費狀態；
- Kafka的設計之初就考慮到了高性能，通過以下方式實現：
  - 利用分區實現並行處理；
  - 使用磁盤順序寫，以及充分利用頁緩存；
  - 零拷貝技術；
  - 批處理技術，數據壓縮等；
RocketMQ：隨着阿里巴巴的電商業務不斷發展，需要一款更高性能的消息中間件，RocketMQ就是這個業務背景的產物。RocketMQ是一個分布式消息中間件，具有低延遲、高性能和可靠性、萬億級別的容量和靈活的可擴展性，它是阿里巴巴於2012年開源的第三代分布式消息中間件。RocketMQ經歷了多年雙十一的洗禮，在可用性、可靠性以及穩定性等方面都有出色的表現。值得一提的是，RocketMQ最初就是借鑒了Kafka進行改造開發而來的，所以熟悉Kafka的朋友，會發現RocketMQ的原理和Kafka有很多相似之處；
Pulsar：在Yahoo，為了追求大集群多租戶、穩定可靠的 IO 服務質量、百萬級 Topic、跨地域復制等需求，Pulsar 應運而生，以彌補Kafka在這方面的不足，Pulsar的優點：
- 應用場景：Pulsar 對用戶來說提供了統一的消息模型，可以滿足各種MQ；
- 架構優勢：有存儲計算分離的雲原生架構的優勢，使用BookKeeper作為Pulsar的存儲層。在 Broker 層不存儲任何數據，具有更高的可用性、更靈活的擴容和管理，避免數據的 reblance 和 catch-up；
- 社區活躍度：Pulsar 用戶和貢獻者數量也在快速增加...

Pub/Sub模式以及第一個消息中間件誕生的故事

1985年在高盛，Ranadive 找到了他的第一個客戶，並確定了他的軟件總線要解決的問題：金融交易。

當時，一個交易員的攤位擠滿了不同的終端，用於提供交易員完成工作所需的每種類型的信息。 Teknekron 看到了替換所有這些終端及其應用程序的機會：通過Ranadive軟件總線取而代之，只需保留一個工作站即可，其顯示程序可以作為消費者插入Teknekron軟件總線，並允許交易者“訂閱”其想要查看的信息。 Pub/Sub 誕生了，世界上第一個現代消息隊列軟件也誕生了：Teknekron的The Information Bus(TIB)。

而RabbitMQ作為傳統的消息中間件，被大量應用於各種古老的項目，你第一個要拿下的就是它了，本文將帶您從以下各個方面了解RabbitMQ相關知識：

什么是AMQP？
常見的交換機類型有哪些？
如何實現消息的持久化？
RabbitMQ的連接復用有啥優勢？
RabbitMQ的消息ACK機制是如何實現的？
RabbitMQ消息持久化機制性能如何？
如何避免消費過載的問題？
如何提高手動ACK簽收的效率？
什么時候需要讓消息重回隊列？
如何保證消息的順序消費？
如何實現可靠的消息投遞？

關於其他的消息中間件，我會在下篇文章中繼續給大家分享。

RabbitMQ是一種使用Erlang語言編寫的開源的消息中間件，最初實現了AMQP(高級消息隊列協議)，后來通過插件架構進行了擴展，支持STOMP(面向流文本的消息傳遞協議)、MQTT(MQ遙控傳輸)等協議。

詳細關於RabbitMQ支持的消息協議，參考官網：Which protocols does RabbitMQ support?^[1]

更多內容歡迎關注公眾號Java架構雜談，或者我的博客IT宅itzhai.com。

1. RabbitMQ優勢

RabbitMQ支持多種客戶端，如Python、Java、.NET、C、Ruby等，在易用性、擴展性、高可用性等方面表現都不錯，並且可以與SpringAMQP完美整合，API豐富易用。

RabbitMQ程序健壯、穩定、易用，跨平台、支持多種語言，管理界面簡單易用，功能全面，文檔相對比較齊全，社區活躍。

2. AMQP簡介

AMQP，全稱為：Advanced Message Queuing Protocol，高級消息隊列協議，是面向消息中間件的開放標准的二進制應用層協議。AMQP的核心特性是：面向消息、排隊、路由(包括點對點和發布訂閱)、可靠性和安全性。這些功能使其非常適合在應用程序之間傳遞業務消息，AMQP還可以用作物聯網IoT協議。

目前，AMQP 1.0已經被批准為國際標准，具體規范文檔，可以進一步閱讀：OASIS Advanced Message Queuing Protocol (AMQP) Version 1.0^[2]

而RabbitMQ最初是為了支持AMQP 0-9-1^[1:1]而開發的，因此，該協議是RabbitMQ Broker支持的核心協議。

下面我們就簡要介紹下AMQP 0-9-1協議^[3]。這部分內容，概念會比較多，稍微有點枯燥，但是可以說RabbitMQ就是按照這個協議去實現的，所以熟悉這個協議很重要。

完整的AMQP文檔可以從這里下載：AMQP Working Group 0-9-1^[4]

2.1 AMQP模型概述

2.1.1 AMQP 0-9-1

AMQP 0-9-1 是一個二進制協議，定義了非常強大的消息傳遞語義。對於客戶端來說，這是一個相對容易實現的協議，因此有大量客戶端庫可用於許多不同的編程語言和環境。

AMQP 0-9-1通常划分為兩層：

功能層(Functional Layer)：定義了一組命令(按功能做不同的分類)，提供給應用程序，用於支撐消息相關的工作；
傳輸層(Transport Layer)：傳輸層將這些方法從應用程序傳送到服務器並返回，並處理通道多路復用、成幀、內容編碼、心跳、數據表示和錯誤處理。

可以在不改變協議的應用程序相關功能的情況下用任意的傳輸協議來替換傳輸層，也可以將傳輸層用於不同的高級協議。

2.1.2 AMQP 0-9-1模型簡介

如下圖，消息Broker(代理)從消息發布者(發布消息的應用程序，也稱為生產者)接收消息並將它們路由到消費者(處理消息的應用程序)。

由於AMQP是一個網絡協議，因此，生產者、消費者、代理都可以部署在不同的機器上。AMQP模型如下圖所示：

消息發布到交換機(exchanges)(通常將其比作郵局或郵箱)，然后使用稱為綁定(Bindings)的規則將消息副本分發到隊列(queues)。然后代理(brokers)要么將消息傳遞(deliver)給訂閱隊列的消費者(consumers)，要么消費者主動按需從隊列中獲取(fetch)/拉取(pull)消息。

消息元數據：發布消息的時候，發布者可以指定各種消息元數據(消息屬性)，其中一些元數據可能由代理使用，其余的元數據僅由接收消息的應用程序使用。

消息確認：由於網絡是不可靠的，並且應用程序可能無法正確處理消息，因此AMQP 0-9-1模型有一個消息確認的概念：當消息傳遞給消費者時，消費者會自動或者由開發人員在應用程序中手動指定通知代理Broker，代理只會在收到消息(或消息組)的通知時從隊列中完全刪除該消息。

死信隊列：在某些情況下，例如，當消息無法路由時，消息可能會返回給發布者、或者丟棄掉、或者將其放入所謂的死信隊列(如果代理擴展支持)，發布者通過使用某些參數來選擇如何處理此類情況。

隊列(queues)/交換機(exchanges)和綁定(bindings)統稱為AMQP實體。

2.1.3 AMQP 0-9-1 是一個可編程的協議

AMQP 0-9-1是一種可編程的協議，因為AMQP 0-9-1實體和路由方案由應用程序本身定義，而不是代理管理員。因此AMQP制定了一些規定來實現這些協議操作：

聲明隊列和交換機；
定義他們之間的綁定；
訂閱隊列等。

這為應用程序開發人員提供了很大的自由，但也要求他們了解潛在的定義沖突。在實踐中，定義沖突很少見，通常表示為配置錯誤。

應用程序聲明它們需要的AMQP 0-9-1實體，定義必要的路由方案，並在不需要使用它們時進行刪除。

2.2 交換機(Exchanges)和交換機類型

交換機是發送消息的AMQP 0-9-1實體。交換機收到一條消息，並將其路由到零個或者多個隊列中。咳咳，Java架構雜談提醒大家，不要聯想到了網絡的交換機(Network switch)，只是中文名稱一樣而已。

使用的路由算法取決於交換機類型和稱為綁定的規則。以下是AMQP 0-9-1 Broker提供的四種交換機類型：

交換類型	默認的預定義名稱
直連交換機(Direct exchange)	空字符串和amq.direct
扇形交換機(Fanout exchange)	amq.fanout
主題交換機(Topic exchange)	amq.topic
頭信息交換機(Headers exchange)	amq.match 和RabbitMQ中的 amq.headers

除了交換類型之外，交換機還聲明了許多屬性，關鍵屬性有：

Name，交換機的名稱
持久性，保證交換機在Broker重啟后仍然存在，如果沒有指定持久，那么交換機在Broker重啟后就不存在了，需要重新聲明，並非所有場景和用例都要求交換機是持久的；
自動刪除，當最后一個隊列解除綁定時，交換機被刪除；
參數，可選，由插件和特定於代理的功能使用。

2.2.1 默認交換機

默認交換機是由Broker預先聲明的匿名直連交換機。使用默認交換機的時候，每個新建隊列都會自動綁定到默認交換機上，綁定的路由鍵與隊列名稱相同，默認交換機看起來可以將消息直接傳遞到隊列。

2.2.2 直連交換機

交換機根據消息路由鍵(router_key)將消息傳遞到隊列，消息將會投遞到與路由鍵名稱和隊列名稱相同的隊列上。直接交換機是消息單播路由的理想選擇(盡管它們也可以用於多播路由)。

直連交換機如下圖所示：

一個隊列N使用路由鍵 K 綁定到交換機；
當具有路由鍵 M 的新消息到達直連交換機時，如果 K = M，則交換機將其路由到隊列N。

如上圖，具有路由鍵"itzhai.com"的消息達到交換機之后，則會路由到Queue1中。

直連交換機通常用於以循環的方式在多個消費者之間分配任務，也就是說，消息的負載均衡是發生在消費者之間而不是隊列之間。

2.2.3 扇形交換機

扇形交換機將消息路由到綁定到它的所有隊列，並且忽略路由鍵。也就是說，當新消息發布到該交換機時，該消息的副本將投遞到所有綁定該交換機的隊列。扇形交換機是消息廣播路由的理想選擇。

扇形交換機如下圖所示：

使用扇形交換機的案例都非常相似：

大型多人在線游戲(MMO)可以將其用於排行榜更新或其他全局事件；
體育新聞網站可以使用扇形交換機向客戶端近乎實時的分發比分信息；
分布式系統使用它來廣播各種狀態和配置更新；
群聊可以使用它在參與者之間分發消息(AMQP沒有內置presence的概念，因此XMPP可能會是更好的選擇)。

2.2.4 主題交換機

主題交換機根據消息路由鍵和和用於將隊列綁定到交換機的模式匹配字符串之間的匹配將消息路由到一個或者多個隊列。

也就是說通過消息的路由鍵去匹配到綁定到交換機的路由鍵匹配字符串，如果匹配上了，就進行投遞消息。

routing key模糊匹配的通配符如下：

*：用於匹配一個單詞，比如 itzhai.com.*，可以匹配：itzhai.com.a，itzhai.com.b
#：用於匹配0個或者多個單詞，比如 itzhai.com.#，可以匹配：itzhai.com.a，itzhai.com.a.b

routing key通過.分隔字符串。

主題交換機如下圖所示：

當生產者發送的routing_key=itzhai.com的時候，會把消息路由投遞到Queue1和Queue2。

當生產者發送的routing_key=www.itzhai.com的時候，會把消息路由投遞到Queue3。

當需要有針對性的選擇多個接收消息的消費者或者應用的時候，主題交換機都可以被列入考慮的范圍。常見的使用場景：

后台任務由多個工作線程完成，每個工作線程負責處理某些特定的任務，這個時候可以通過主題交換機訂閱感興趣的消息；
分發特定地理位置的信息，每個工作線程只訂閱感興趣的地理位置的信息；
...

2.2.5 頭交換機

頭交換機不依賴路由鍵的匹配規則來路由消息，而是根據發送消息內容中的請求頭屬性進行匹配。

頭交換機類似於直連交換機，但是直連交換機的路由鍵必須是一個字符串，而請求頭信息則沒有這個約束，它們甚至可以是整數或者字典。因此可以用作路由鍵不必是字符串的直連交換。

綁定一個隊列到頭交換機上的時候，會同時綁定多個用於匹配的頭信息。

投遞消息的時候，可以攜帶一個x-match參數，指定是否要求必須所有的頭信息都匹配(all)才能進行投遞，還是只要匹配任何一個就可以了(any)。

注意：以字符串x-打頭的頭屬性，不會作為匹配項。

2.3 隊列(Queues)

AMQP 0-9-1 中的隊列與其他消息隊列和任務隊列系統中的隊列類似，用於存儲即將被消費的消息。一般地，隊列與交換機共享一些屬性，但隊列也有一些特定的屬性：

Name：隊列名稱；
Durable：隊列持久化，隊列在Broker重啟之后是否繼續存在；
Exclusive：隊列是否僅由一個連接使用，如果是，在連接關閉的時候，隊列將被刪除；
Auto-delete：當最后一個消費者取消訂閱的時候，立即刪除；
Arguments：可選，一些特定的插件和Broker功能使用，例如實現消息的TTL，隊列長度限制等。

關於隊列的聲明：

在使用隊列之前，必須先聲明它。聲明隊列的時候，如果隊列尚不存在，則將創建一個隊列；如果隊列已存在，並且屬性與聲明中的屬性相同，則不用重新創建一個；如果現有隊列屬性與聲明的隊列屬性不同，將會引發406(PRECONDITION_FAILED)的通道級異常。

2.3.1 隊列名稱

應用程序可以設置隊列名稱，如果設置為空字符串，Broker會為它們生成一個唯一的隊列名稱，在隊列聲明響應體中一起返回給客戶端。隊列名稱為255個字節以內的UTF-8字符。

以amq開頭的隊列名稱，保留給Broker內部使用，如果嘗試使用此類名稱聲明一個隊列將導致通道級別異常：403(ACCESS_REFUSED)。

2.3.2 隊列持久化

持久化的隊列的元數據會存儲在磁盤上，當Broker重啟之后，隊列依然存在。沒有被持久化的隊列稱為暫存隊列。發布的消息也有同樣的區分，也就是說，持久化的隊列並不會使得路由到它的消息也具有持久性，需要手動把消息也標記為持久化才能保證消息的持久性。

2.4 綁定(Bindings)

綁定是交換機將消息路由到隊列的規則。為了讓交換機能夠正確的把消息投遞到對應的隊列，需要把交換機和隊列通過路由鍵綁定起來，路由鍵就像是一個過濾器，決定了消息是否可以投遞給消息隊列。

如果一條消息不能被路由到任何隊列(例如，因為它被發布到的交換機沒有綁定)，它要么被丟棄，要么返回給發布者，這取決於發布者設置的消息屬性。

2.5 消費者

如果消息只是存儲在隊列里沒有被消費，是沒有什么實際作用的。在AMQP 0-9-1中，有兩種途徑可以進行消息的消費：

訂閱消息隊列，以將消息投遞給應用(push API)，這是推薦的做法；
應用根據需要主動的輪訓獲取消息(pull API)，這種方式非常低效，在大多數情況下應該避免。

如果應用程序對某一個特定隊列的消息感興趣，則可以注冊一個消費者，對隊列進行訂閱。每個隊列可以有多個消費者，當然也可以注冊一個獨享的消費者，這個時候其他消費者會被排除在外。

每個消費者(訂閱)都有一個稱為消費者標簽的字符串類型的標識符，可以用它來退訂消息。

2.5.1 消息確認

消費者應用程序可能偶爾無法處理單個消息或有時會崩潰，另外網絡問題也有可能導致問題。這就提出了一個問題：Broker何時應該從隊列中刪除消息？AMQP 0-9-1 規范中約定讓消費者對此進行控制，有兩種確認模式：

自動確認模式：在Broker向應用程序發送消息之后(使用basic.deliver或basic.get-ok方法)，將消息從消息隊列中刪除；
顯示確認模式：在應用程序向broker發回確認之后(使用basic.ack方法)，將消息從消息隊列中刪除。

在顯示模式下，應用程序選擇何時發送確認消息。如果消費者在沒有發送確認的情況下就掛掉了，那么Broker會將其重新投遞給另一個消費者，如果此時沒有可用的消費者，那么Broker將等到至少有一個消費者注冊到該隊列時，再嘗試重新投遞消息。

另外，如果應用程序崩潰(當連接關閉時 AMQP Broker會感知到這一點)，並且AMQP Broker在預期的時間內未收到消息確認，則消息將重新入隊，如果此時有其他消費者，可能立即傳遞給另一個消費者。為此，我們的消費者做好業務的冪等處理也是非常重要的。

2.5.2 拒絕消息

當消費者接收到消息之后，可能處理成功或者失敗。應用程序可以通過拒絕消息向Broker表明消息處理失敗了(或者當時無法完成)。拒絕消息的時候，應用程序可以要求Broker丟棄消息或者重新入隊。

當隊列中只有一個消費者的時候，請確保您不會通過不斷地拒絕消息和重新入隊導致消息在同一個消費者身上無限循環的情況發生。

在AMQP中，basic.reject方法用來執行拒絕消息的操作。

2.5.3 預取消息

在多個消費者共享一個隊列的情況，能夠制定每個消費者在發送下一個ack之前可以一次性接收多少條消息，這是非常有用的特性。這可以在試圖批量發布消息的時候，起到簡單的負載均衡和提高消息吞吐量的作用。

請注意：RabbitMQ僅支持通道級預取計數，不支持基於連接或者大小的預取。

2.6 消息屬性和有效負載

AMQP 0-9-1模型中的消息是具有屬性的，有些屬性非常常見，以至於AMQP 0-9-1明確定義了它們，例如：

Content type 內容類型
Content encoding 內容編碼
Routing key 路由鍵
Delivery mode (persistent or not) 投遞模式，是否持久化
Message priority 消息優先級
Message publishing timestamp 消息發布的時間戳
Expiration period 消息有效期
Publisher application id 發布消息的應用id

有些屬性是被AMQP的Broker所使用的，但是大多數是開放給接收它們的應用程序用的。有些屬性是可選的，稱為消息頭（headers），它們類似於HTTP協議的X-Headers，消息屬性需要在消息被發布時定義。

消息體：AMQP消息除了屬性之外，還包括一個有效載荷Payload(消息實際攜帶的數據)，AMQP Broker視其為一個透明的字節數組來對待。Broker不會修改payload。消息可能只包含屬性而沒有payload。payload通常使用JSON、Thrift、Protocol Buffers和MessagePack等序列化格式來序列化成結構化的數據，以便進行發布，協議對等方通常使用Content type和Content encoding字段來傳達此信息。

消息持久化：消息可以作為持久性發布，這使得Broker將他們持久化到磁盤。如果服務器重啟之后，系統可以確保接收到的持久化消息不會丟失。簡單的將消息發布到持久化的交換機或者被路由到持久化的隊列中，是不會讓消息也持久化的，消息是否持久化完全取決於消息本身的持久模式。將消息發布為持久性會影響性能，就像數據存儲一樣，持久性以一定的性能成本作為代價。

2.7 AMQP 0-9-1 方法

AMQP 0-9-1中定義了許多操作方法，詳細參考：AMQP 0-9-1參考^[5]。

很多方法都有對應的響應方法，有些甚至有不止一種可能的響應，如basic.get，響應可能為：get-ok或者get-empty。

如下是聲明一個交換機和響應成功的方法：

2.8 連接(Connections)

AMQP 0-9-1 連接通常是長連接，AMQP 0-9-1 是一種使用TCP提供可靠投遞的應用層協議。AMQP 0-9-1連接使用身份認證機制並提供TLS (SSL)保護。當應用程序不再需要連接到Broker時，它應該優雅地關閉其 AMQP 0-9-1 連接，而不是突然關閉底層 TCP 連接。

2.9 通道(Channels)

某些應用程序需要同時開啟連接到Broker，但是，同時保持許多TCP連接是不可取的，這樣會消耗系統資源並且使得配置防火牆更加困難。

AMQP 0-9-1通過通道復用技術通過通道的形式實現在一個TCP連接上面支持多個連接(虛擬的鏈接)。同一個TCP連接中有多個通道，通道之間的通信是完全隔離的。客戶端的每個協議操作都攜帶了一個通道ID，代理和客戶端都是用它來確定該操作所走的通道。

通道僅存在於TCP連接上下文中，一旦TCP連接關閉，其上所有通道也跟着關閉。

一般的，我們會給每個線程打開一個新的通道進行通信。

2.10 虛擬主機

為了讓單個代理可以托管多個隔離的環境(用戶組、交換機、隊列等)，AMQP中提供了虛擬主機，這類似於許多流行的Web服務器使用的虛擬主機。協議客戶端在連接協商期間需要指定想要使用的虛擬主機。

2.11. AMQP Client架構

推薦的AQMP Client架構須由下面多個抽象層組成：

成幀層：此層接收AMQP協議方法，並按某種語言格式(結構、類等)來序列化成線級幀，成幀層可以根據AMQP協議規范來實現；
連接管理層：此層用於讀寫AMQP幀，並管理所有連接、會話邏輯。在此層中，我們可以封裝開啟連接和會話、錯誤處理、內容傳輸和接收數據的全部邏輯；
API層：此層暴露了應用程序工作的特定API。API層可能會反映一些現有的標准，或暴露高層AMQP的方法。API層本身可能是由多個層組成的，如構建於AMQP方法API之上的高級API；
IO層：此外，通常還會有一些I/O層，這此可以是非常簡單的同步套接字讀取和寫入或復雜的異步多線程IO。

AMQP就介紹到這里了，接下來Java架構雜談帶大家詳細看看RabbitMQ。

3. RabbitMQ架構

RabbitMQ的整體架構如下圖所示：

Broker：Broker中按虛擬主機(virtual host)划分，每個虛擬主機下面有自己的交換機(exchange)和消息隊列(queue)，以及交換機和隊列的綁定routing_key(有些人會把這個key稱為binding_key)；

生產端：一般地，同一個客戶端(client)里面的每個生產者(producer)創建一個專門的通道(channel)，復用同一個TCP連接(connection)，每個生產者可以往Broker發布消息，發布消息的時候，需指定虛擬主機，以及虛擬主機上的交換機，並且消息需要帶上routing_key；

消費端：一般地，同一個客戶端(client)里面的每個消費者(consumer)創建一個專門的通道(channel)，復用同一個TCP連接，每個消費者指定一個消息隊列進行消費。同一個消息隊列，可以有多個消費者共同消費，但消息隊列里面的同一條消息，只會由一個消費者消費，多個消費者相當於給消息隊列做了負載均衡。

針對默認交換機、直連交換機和主題交換機，生產端帶入的routing_key和交換機與隊列之間綁定的routing_key(binding_key)進行匹配，匹配上了，就把消息投遞給對應的消息隊列。

針對扇形交換機，直接把消息投遞給所有與扇形交換機綁定的隊列。

rabbitmqctl是管理RabbitMQ服務器節點的主要命令行工具，相關完成命令介紹參考：rabbitmqctl(8)^[6]

4. RabbitMQ特性

4.1 消息ACK機制^[7]

ACK (Acknowledge character)，即是確認字符，消息的接收方需要告訴發送方已確認接收消息，這是實現可靠消息投遞的必備特性。

MQ系統中，涉及到ACK的流程如下圖所示：

4.1.1 生產端ACK之Confirm消息機制

如上圖所示：

Producer發布消息到Broker；
Broker將消息落地；
Broker發送ack給Producer。

如果Producer沒有收到ack，那么可以重發消息，直到收到ack為止。為了避免無限的給Broker投遞消息，應該設置一個重試上限，並記錄下發送失敗的消息。在這個過程中，MQ Server可能會收到重復消息。

在RabbitMQ中，生產端的ACK通過ConfirmListener機制來實現：

在channel中開啟確認模式confirmSelect()，然后在channel中添加監聽，用來監聽Broker返回的應答。

Broker何時給生產端發送ACK？

對於不可路由的消息，一旦交換機驗證消息不會路由到任何隊列，Broker將發出ack，如果開啟了Return消息機制(下一小節講解)，那么Broker會先發送basic.return消息給客戶端，再發送basic.ack消息。示例代碼如下：

String message = "Hello itzhai.com....";
// Confirm消息機制
channel.addConfirmListener(new TestConfirmListener());

// Return消息機制
channel.addReturnListener(new TestRetrunListener());
// 錯誤的路由鍵，但交換機的名稱正確
String errorRoutingKey = "itzhai.com.test1";
boolean mandatory = true;

channel.basicPublish(exchangeName, errorRoutingKey, mandatory, basicProperties, message.getBytes());

執行以上代碼，生產者將依次收到basic.return(Return消息)，basic.ack(Confirm消息)。

對於可路由的消息，當所有隊列都接收到消息的之后，Broker向生產端發送ACK。如果路由到的是持久隊列，並且是持久消息，那么這個ACK就意味着消息持久化到了磁盤。

也就是說，路由到持久隊列的持久消息的ACK將在將消息持久化到磁盤后發送。

RabbitMQ消息持久化的性能如何？

RabbitMQ持久化消息的刷盤策略：為了盡可能減少fsync(2)的調用次數，RabbitMQ在間隔一段時間(幾百毫秒)或者在隊列空閑的時候將消息分批保存到磁盤中。

這就意味着，在正常的負載下，生產端接收Broker的ACK時延可達幾百毫秒。為了提高吞吐量，強烈建議生產端應用程序異步處理ACK，或者批量發布消息，並等待ACK。

4.1.2 生產端ACK之Return消息機制

Return消息機制用於處理一些不可路由的消息。發送消息的時候，如果指定的routing_key路由不到隊列，這個時候就可以通過ReturnListener監聽這種異常情況。

4.1.3 消費端ACK

如上圖所示：

消息服務器將消息投遞給消費者；
消費者消費消息，並向消息服務器發送ack；
消息服務器收到消費者的ack之后，將已落地的消息刪除掉。

當Broker一直沒有收到消費端的ACK，則會重發消息，這個過程一般采用指數退避策略，時間間隔按指數增長。

Rabbit中的消費端ACK

在RabbitMQ中，消費端的ACK可以是自動的，或者手動的。

手動ACK簽收

通過以下方法關閉自動ack簽收(入參autoAck設置為false)：

Channel.java
String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;

然后自定義一個支持ack的Consumer：

public class TestAckConsumer extends DefaultConsumer {

    ...

    public void handleDelivery(String consumerTag,
                               Envelope envelope,
                               AMQP.BasicProperties properties,
                               byte[] body)
        throws IOException
    {
        try{
            ...
            // 成功消費的ack
            boolean multiple = false;
            channel.basicAck(envelope.getDeliveryTag(), multiple);
        }catch (Exception e) {
            // 未成功消費的ack，設置為不重回隊列，即立刻刪除消息
            boolean multiple = false;
            boolean requeue = false;
            channel.basicNack(envelope.getDeliveryTag(), multiple, requeue);
        }
    }
}

channel中有三種ack相關的方法：

basic.ack：用於肯定確認，指示RabbitMQ消息已經處理成功，可以丟棄消息了；
basic.nack：用於否定確認，指示RabbitMQ消息未處理成功，可以通過參數指定是否需要丟棄消息還是重回隊列。
basic.reject：用於否定確認，指示RabbitMQ消息未處理成功，可以通過參數指定是否需要丟棄消息還是重回隊列。

basic.nack與basic.reject的區別就是，basic.nack支持批量手動確認，basic.nack是RabbitMQ對AMQP 0-9-1協議的擴展。

自動ACK簽收

使用自動確認模式，消息在發送之后就立刻被標記為投遞消費成功。如果消費者的TCP連接或者通道在真正投遞成功之前就關閉了，那么Broker發送的消息將會丟失。自動確認模式是以降低消息投遞的可靠性來換取更高的消費端吞吐量(只要消費端處理速度能夠跟上)。

如何避免消費過載的問題(消費端限流)？

使用自動模式可以提高吞吐量，但是前提是消費端要能夠處理得過來，如果處理不過來，就會在消費端的內存中積壓消息，直至把內存耗盡。因此，自動確認模式僅推薦用於能夠以穩定的速度高效地處理消息的消費者。

為了避免消費過載問題，我們一般使用手動確認模式，配合通道預取限制一起使用：

// 每條消息的大小限制，0表示不限制
int prefetchSize = 0;  
// MQ Server每次推送的消息的最大條數，0表示不限制
int prefetchCount = 1;  
// true 表示配置應用於整個通道，false表示只應用於消費級別
boolean global = false;  
channel.basicQos(prefetchSize, prefetchCount, global);

// 隊列名稱
String queueName = "com.itzhai.queue";  
// 設置為手動確認模式
boolean autoAck = false;
// 消費者對象實例
Consumer consumer = new ItzhaiTestConsumer(channel);
channel.basicConsume(queueName, autoAck, consumer);

如何提高手動ACK簽收的效率

如果不需要嚴格控制發送消費端ACK的時間，即，只要消費者成功接收到消息，不管有沒有消費成功，都允許進行ACK回復，那么就可以通過批量ACK簽收的功能更來提高簽收的消效率。做法如下：

// 手動簽收模式
boolean autoAck = false;
channel.basicConsume(queueName, autoAck, "a-consumer-tag",
     new DefaultConsumer(channel) {
         @Override
         public void handleDelivery(String consumerTag,
                                    Envelope envelope,
                                    AMQP.BasicProperties properties,
                                    byte[] body)
             throws IOException
         {
             long deliveryTag = envelope.getDeliveryTag();
             // 注意，這里設置為批量簽收
             boolean mutiple = true;
             // 簽收deliveryTag以及deliveryTag之前的所有消息
             channel.basicAck(deliveryTag, mutiple);
         }
     });

這樣執行basicAck，deliveryTag以及deliveryTag之前的所有消息都將會被簽收。

什么時候需要讓消息重回隊列？

有時候消費者太繁忙導致無法立即處理接收到的消息，但是其他實例可能可以處理。這種情況，就可以拒絕消息，並且讓消息重回隊列。

另外，可以使用channel.basicNack方法一次拒絕或者重新排隊多條消息：

// 指定批量拒絕策略
boolean multiple = true;
// 指定拒絕之后重新入隊
boolean requeue = true;
channel.basicNack(envelope.getDeliveryTag(), multiple, requeue);

極端情況下，如果所有消費者因為暫時無法處理接收的消息，會導致消息不斷的循環重回入隊，導致消耗網絡帶寬和CPU資源。為了避免這種情況，可以跟蹤重回隊列的消息數量，決定是否需要永久拒絕消息(丟棄消息)還是延遲重回隊列的時間。

4.2 消息的順序性能夠得到保證嗎？

一般情況下，在單個通道上發布的消息，Rabbit會按照消息發布的相通順序向生產端發送ACK消息，但也不是絕對的。發布ACK的確切時刻取決於消息的傳遞模式(持久化或瞬時)，以及消息路由到的隊列的屬性。也就是說，不同的消息在不同的時間准備好進行確認，確認消息可以以不同的順序達到。所以，應用程序盡可能不要依賴於消息的順序性。

4.3 消息處理的冪等性如何處理？

無論是生產端還是消費端的ACK，都有可能因為網絡或者程序問題導致ACK消息沒有及時送達，這個時候會導致重復的消息投遞。如何保證消費同一條消息的情況下不影響業務，這就需要保證消息處理的冪等性。

也就是說，針對同一條消息，無論消費者消費多少次，產生的效果始終應該跟消費一次的保持一致，並且返回的ACK結果也是一致的。

常用的實現消息處理冪等性的方法：

每條消息生成唯一ID，消費端根據唯一ID判斷是否已經消費過，如果消費過，則直接返回消費成功的ACK。
- 針對入庫的業務操作可以通過數據庫的唯一索引來實現避免重復業務數據入庫；
- 針對修改數據類的操作，可以先判斷數據是否已經是目標狀態了，如果是目標狀態，直接返回再進行更新。
針對並發的場景，我們需要給業務消費程序添加分布式鎖，避免並發執行導致觸發業務重復處理。

4.4 死信隊列^[8]

如果消息隊列中的消息沒有被正常消費掉，那么該消息就會成為一個死信(Dead Letter)，這條消息可以被重新發送到另一個交換機上，后面這個交換機就是死信交換機(DLX)，死信交換機綁定的隊列就是死信隊列。在以下情況下導致的消息未被正常消費，均會使消息變為死信：

消費者使用basic.reject或者basic.nack來拒絕消息，同時設置requeue參數為false，表示消息不需要重回隊列；
消息設置了TTL，並且過期了，或者隊列設置了消息的過期時間x-message-ttl；
由於消息隊列超過了長度限制導致消息被丟棄了。

死信隊列也是一個正常的交換機，它可以是任何常見的交換機類型，與常規交換機聲明沒有區別。

DLX可以有客戶端使用隊列參數(arguments)進行定義，或者在服務器中使用策略(policy)進行定義，在policy和arguments都定義了的情況下，arguments中指定的那個會否決policy中指定的那個。

通過policy啟用死信隊列：

rabbitmqctl set_policy DLX ".*" '{"dead-letter-exchange":"my-dlx"}' --apply-to queues

通過arguments啟用死信隊列：

// 聲明一個交換機，作為死信交換機
channel.exchangeDeclare("some.exchange.name", "direct");

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-dead-letter-exchange", "some.exchange.name");
channel.queueDeclare("myqueue", false, false, false, args);

5. 持久化消息就意味着消息的可靠性嗎？如何實現可靠性投遞？

消息可靠性需要考慮生產端投遞消息的可靠性以及保證消費端最終成功地消費消息。

雖然通過生產端的ACK機制，可以確保消息成功的投遞到了RabbitMQ中，保證投遞的消息不丟失。但是如果生產端不知道消費者究竟有沒有成功的消費了消息，那也就無法實現可靠性投遞了。

而生產端投遞消息的過程中，通常會涉及到生產端的事務提交，要保證消息跟隨事務提交而發送，也是需要考慮的問題。

如何實現可靠投遞呢？這里留給大家思考，關鍵設計要點：

是否要發消息跟隨生產端事務一起保存到發送日志表，提交事務之后立刻向消息隊列投遞一次消息；
- 生產端發送日志表消息狀態：1 發送中，2 Broker簽收成功，3 Broker簽收失敗，4 消費端簽收成功，5 消費端簽收失敗
使用消息隊列模擬RPC調用，在消費者成功處理消息之后，向生產者投遞成功消費的消息，以便讓生產端知道消息已經處理成功了；
定時任務定時掃描生產端發送日志表，對於超過固定時間之內，還未處理成功的消息，進行重試投遞，重試可以使用指數退避策略，並設置投遞上限次數。如果達到上限次數還未成功，則預警人工介入排查；
消費端一定要做好冪等處理，避免重復消費導致業務異常。

提示的還不夠具體？我再上一張圖：

有更好的方案的朋友，歡迎在評論區留言交流，也許你就是評論區最靚的仔。

6. RabbitMQ更多使用場景

通過給消息設置TTL，超時時候放入死信隊列進行處理，可以實現延遲隊列，當然，RabbitMQ也有專門的延遲隊列插件可以使用；

另外，也可以使用RabbitMQ模擬RPC調用，參考上一節實現消息可靠性投遞的例子。

更多的使用場景歡迎大家進行補充。

關於更多消息中間件的文章，歡迎關注Java架構雜談，或者我的博客IT宅(itzhai.com)，我會持續的輸出相關內容。

我精心整理了一份Redis寶典給大家，涵蓋了Redis的方方面面，面試官懂的里面有，面試官不懂的里面也有，有了它，不怕面試官連環問，就怕面試官一上來就問你Redis的Redo Log是干啥的？畢竟這種問題我也不會。

在Java架構雜談公眾號發送Redis關鍵字獲取pdf文件：

本文作者： arthinking

博客鏈接： https://www.itzhai.com/articles/rabbitmq-advanced-tutorial.html

消息隊列那么多，為什么建議深入了解下RabbitMQ？

版權聲明：版權歸作者所有，未經許可不得轉載，侵權必究！聯系作者請加公眾號。

References

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 深入了解RabbitMQ RabbitMQ系列（二）深入了解RabbitMQ工作原理及簡單使用從網絡I/O模型到Netty，先深入了解下I/O多路復用 RabbitMQ入門_06_深入了解ack 深入了解mitmproxy（二）深入了解final 深入了解ANR 深入了解ALSA MVVM深入了解深入了解JVW