面渣逆襲：RocketMQ二十三問

基礎

1.為什麼要使用消息隊列呢？

消息隊列主要有三大用途，我們拿一個電商系統的下單舉例：

  • 解耦：引入消息隊列之前，下單完成之後，需要訂單服務去調用庫存服務減庫存，調用營銷服務加營銷數據……引入消息隊列之後，可以把訂單完成的消息丟進隊列里，下游服務自己去調用就行了，這樣就完成了訂單服務和其它服務的解耦合。

    消息隊列解耦

  • 非同步：訂單支付之後，我們要扣減庫存、增加積分、發送消息等等，這樣一來這個鏈路就長了，鏈路一長，響應時間就變長了。引入消息隊列，除了更新訂單狀態，其它的都可以非同步去做，這樣一來就來，就能降低響應時間。

    消息隊列非同步

  • 削峰：消息隊列合一用來削峰，例如秒殺系統，平時流量很低，但是要做秒殺活動，秒殺的時候流量瘋狂懟進來，我們的伺服器，Redis，MySQL各自的承受能力都不一樣，直接全部流量照單全收肯定有問題啊，嚴重點可能直接打掛了。

    我們可以把請求扔到隊列裡面，只放出我們服務能處理的流量，這樣就能抗住短時間的大流量了。

    消息隊列削峰

    解耦、非同步、削峰，是消息隊列最主要的三大作用。

2.為什麼要選擇RocketMQ?

市場上幾大消息隊列對比如下：

四大消息隊列對比

總結一下：

選擇中間件的可以從這些維度來考慮：可靠性，性能，功能，可運維行，可拓展性，社區活躍度。目前常用的幾個中間件，ActiveMQ作為「老古董」，市面上用的已經不多，其它幾種：

  • RabbitMQ：

  • 優點：輕量，迅捷，容易部署和使用，擁有靈活的路由配置

  • 缺點：性能和吞吐量不太理想，不易進行二次開發

  • RocketMQ：

    • 優點：性能好，高吞吐量，穩定可靠，有活躍的中文社區

    • 缺點：兼容性上不是太好

  • Kafka：

    • 優點：擁有強大的性能及吞吐量，兼容性很好
    • 缺點：由於「攢一波再處理」導致延遲比較高

我們的系統是面向用戶的C端系統，具有一定的並發量，對性能也有比較高的要求，所以選擇了低延遲、吞吐量比較高，可用性比較好的RocketMQ。

3.RocketMQ有什麼優缺點？

RocketMQ優點：

  • 單機吞吐量：十萬級
  • 可用性：非常高，分散式架構
  • 消息可靠性：經過參數優化配置，消息可以做到0丟失
  • 功能支援：MQ功能較為完善，還是分散式的，擴展性好
  • 支援10億級別的消息堆積，不會因為堆積導致性能下降
  • 源碼是Java，方便結合公司自己的業務二次開發
  • 天生為金融互聯網領域而生，對於可靠性要求很高的場景，尤其是電商裡面的訂單扣款，以及業務削峰，在大量交易湧入時，後端可能無法及時處理的情況
  • RoketMQ在穩定性上可能更值得信賴，這些業務場景在阿里雙11已經經歷了多次考驗，如果你的業務有上述並發場景，建議可以選擇RocketMQ

RocketMQ缺點：

  • 支援的客戶端語言不多，目前是Java及c++，其中c++不成熟
  • 沒有在 MQ核心中去實現JMS等介面，有些系統要遷移需要修改大量程式碼

4.消息隊列有哪些消息模型？

消息隊列有兩種模型：隊列模型和發布/訂閱模型。

  • 隊列模型

    這是最初的一種消息隊列模型，對應著消息隊列「發-存-收」的模型。生產者往某個隊列裡面發送消息，一個隊列可以存儲多個生產者的消息，一個隊列也可以有多個消費者，但是消費者之間是競爭關係，也就是說每條消息只能被一個消費者消費。

    隊列模型

  • 發布/訂閱模型

    如果需要將一份消息數據分發給多個消費者，並且每個消費者都要求收到全量的消息。很顯然，隊列模型無法滿足這個需求。解決的方式就是發布/訂閱模型。

    在發布 – 訂閱模型中，消息的發送方稱為發布者（Publisher），消息的接收方稱為訂閱者（Subscriber），服務端存放消息的容器稱為主題（Topic）。發布者將消息發送到主題中，訂閱者在接收消息之前需要先「訂閱主題」。「訂閱」在這裡既是一個動作，同時還可以認為是主題在消費時的一個邏輯副本，每份訂閱中，訂閱者都可以接收到主題的所有消息。

    發布-訂閱模型

    它和 「隊列模式」 的異同：生產者就是發布者，隊列就是主題，消費者就是訂閱者，無本質區別。唯一的不同點在於：一份消息數據是否可以被多次消費。

5.那RocketMQ的消息模型呢？

RocketMQ使用的消息模型是標準的發布-訂閱模型，在RocketMQ的術語表中，生產者、消費者和主題，與發布-訂閱模型中的概念是完全一樣的。

RocketMQ本身的消息是由下面幾部分組成：

RocketMQ消息的組成

  • Message

Message（消息）就是要傳輸的資訊。

一條消息必須有一個主題（Topic），主題可以看做是你的信件要郵寄的地址。

一條消息也可以擁有一個可選的標籤（Tag）和額處的鍵值對，它們可以用於設置一個業務 Key 並在 Broker 上查找此消息以便在開發期間查找問題。

  • Topic

Topic（主題）可以看做消息的歸類，它是消息的第一級類型。比如一個電商系統可以分為：交易消息、物流消息等，一條消息必須有一個 Topic 。

Topic 與生產者和消費者的關係非常鬆散，一個 Topic 可以有0個、1個、多個生產者向其發送消息，一個生產者也可以同時向不同的 Topic 發送消息。

一個 Topic 也可以被 0個、1個、多個消費者訂閱。

  • Tag

Tag（標籤）可以看作子主題，它是消息的第二級類型，用於為用戶提供額外的靈活性。使用標籤，同一業務模組不同目的的消息就可以用相同 Topic 而不同的 Tag 來標識。比如交易消息又可以分為：交易創建消息、交易完成消息等，一條消息可以沒有 Tag 。

標籤有助於保持你的程式碼乾淨和連貫，並且還可以為 RocketMQ 提供的查詢系統提供幫助。

  • Group

RocketMQ中，訂閱者的概念是通過消費組（Consumer Group）來體現的。每個消費組都消費主題中一份完整的消息，不同消費組之間消費進度彼此不受影響，也就是說，一條消息被Consumer Group1消費過，也會再給Consumer Group2消費。

消費組中包含多個消費者，同一個組內的消費者是競爭消費的關係，每個消費者負責消費組內的一部分消息。默認情況，如果一條消息被消費者Consumer1消費了，那同組的其他消費者就不會再收到這條消息。

  • Message Queue

Message Queue（消息隊列），一個 Topic 下可以設置多個消息隊列，Topic 包括多個 Message Queue ，如果一個 Consumer 需要獲取 Topic下所有的消息，就要遍歷所有的 Message Queue。

RocketMQ還有一些其它的Queue——例如ConsumerQueue。

  • Offset

在Topic的消費過程中，由於消息需要被不同的組進行多次消費，所以消費完的消息並不會立即被刪除，這就需要RocketMQ為每個消費組在每個隊列上維護一個消費位置（Consumer Offset），這個位置之前的消息都被消費過，之後的消息都沒有被消費過，每成功消費一條消息，消費位置就加一。

也可以這麼說，Queue 是一個長度無限的數組，Offset 就是下標。

RocketMQ的消息模型中，這些就是比較關鍵的概念了。畫張圖總結一下：
RocketMQ消息模型

6.消息的消費模式了解嗎？

消息消費模式有兩種：Clustering（集群消費）和Broadcasting（廣播消費）。

兩種消費模式

默認情況下就是集群消費，這種模式下一個消費者組共同消費一個主題的多個隊列，一個隊列只會被一個消費者消費，如果某個消費者掛掉，分組內其它消費者會接替掛掉的消費者繼續消費。

而廣播消費消息會發給消費者組中的每一個消費者進行消費。

7.RoctetMQ基本架構了解嗎？

先看圖，RocketMQ的基本架構：

RocketMQ架構

RocketMQ 一共有四個部分組成：NameServer，Broker，Producer 生產者，Consumer 消費者，它們對應了：發現、發、存、收，為了保證高可用，一般每一部分都是集群部署的。

8.那能介紹一下這四部分嗎？

類比一下我們生活的郵政系統——

郵政系統要正常運行，離不開下面這四個角色， 一是發信者，二 是收信者， 三是負責暫存傳輸的郵局， 四是負責協調各個地方郵局的管理機構。 對應到 RocketMQ 中，這四個角色就是 Producer、 Consumer、 Broker 、NameServer。

RocketMQ類比郵政體系

NameServer

NameServer 是一個無狀態的伺服器，角色類似於 Kafka使用的 Zookeeper，但比 Zookeeper 更輕量。
特點：

  • 每個 NameServer 結點之間是相互獨立，彼此沒有任何資訊交互。
  • Nameserver 被設計成幾乎是無狀態的，通過部署多個結點來標識自己是一個偽集群，Producer 在發送消息前從 NameServer 中獲取 Topic 的路由資訊也就是發往哪個 Broker，Consumer 也會定時從 NameServer 獲取 Topic 的路由資訊，Broker 在啟動時會向 NameServer 註冊，並定時進行心跳連接，且定時同步維護的 Topic 到 NameServer。

功能主要有兩個：

  • 1、和Broker 結點保持長連接。
  • 2、維護 Topic 的路由資訊。

Broker

消息存儲和中轉角色，負責存儲和轉發消息。

  • Broker 內部維護著一個個 Consumer Queue，用來存儲消息的索引，真正存儲消息的地方是 CommitLog（日誌文件）。

RocketMQ存儲-圖片來源官網

  • 單個 Broker 與所有的 Nameserver 保持著長連接和心跳，並會定時將 Topic 資訊同步到 NameServer，和 NameServer 的通訊底層是通過 Netty 實現的。

Producer

消息生產者，業務端負責發送消息，由用戶自行實現和分散式部署。

  • Producer由用戶進行分散式部署，消息由Producer通過多種負載均衡模式發送到Broker集群，發送低延時，支援快速失敗。
  • RocketMQ 提供了三種方式發送消息：同步、非同步和單向
    • 同步發送：同步發送指消息發送方發出數據後會在收到接收方發迴響應之後才發下一個數據包。一般用於重要通知消息，例如重要通知郵件、營銷簡訊。
    • 非同步發送：非同步發送指發送方發出數據後，不等接收方發迴響應，接著發送下個數據包，一般用於可能鏈路耗時較長而對響應時間敏感的業務場景，例如用戶影片上傳後通知啟動轉碼服務。
    • 單向發送：單向發送是指只負責發送消息而不等待伺服器回應且沒有回調函數觸發，適用於某些耗時非常短但對可靠性要求並不高的場景，例如日誌收集。

Consumer

消息消費者，負責消費消息，一般是後台系統負責非同步消費。

  • Consumer也由用戶部署，支援PUSH和PULL兩種消費模式，支援集群消費和廣播消費，提供實時的消息訂閱機制。
  • Pull：拉取型消費者（Pull Consumer）主動從消息伺服器拉取資訊，只要批量拉取到消息，用戶應用就會啟動消費過程，所以 Pull 稱為主動消費型。
  • Push：推送型消費者（Push Consumer）封裝了消息的拉取、消費進度和其他的內部維護工作，將消息到達時執行的回調介面留給用戶應用程式來實現。所以 Push 稱為被動消費類型，但其實從實現上看還是從消息伺服器中拉取消息，不同於 Pull 的是 Push 首先要註冊消費監聽器，當監聽器處觸發後才開始消費消息。

進階

9.如何保證消息的可用性/可靠性/不丟失呢？

消息可能在哪些階段丟失呢？可能會在這三個階段發生丟失：生產階段、存儲階段、消費階段。

所以要從這三個階段考慮：

消息傳遞三階段

生產

在生產階段，主要通過請求確認機制，來保證消息的可靠傳遞。

  • 1、同步發送的時候，要注意處理響應結果和異常。如果返迴響應OK，表示消息成功發送到了Broker，如果響應失敗，或者發生其它異常，都應該重試。
  • 2、非同步發送的時候，應該在回調方法里檢查，如果發送失敗或者異常，都應該進行重試。
  • 3、如果發生超時的情況，也可以通過查詢日誌的API，來檢查是否在Broker存儲成功。

存儲

存儲階段，可以通過配置可靠性優先的 Broker 參數來避免因為宕機丟消息，簡單說就是可靠性優先的場景都應該使用同步。

  • 1、消息只要持久化到CommitLog（日誌文件）中，即使Broker宕機，未消費的消息也能重新恢復再消費。
  • 2、Broker的刷盤機制：同步刷盤和非同步刷盤，不管哪種刷盤都可以保證消息一定存儲在pagecache中（記憶體中），但是同步刷盤更可靠，它是Producer發送消息後等數據持久化到磁碟之後再返迴響應給Producer。

同步刷盤和非同步刷盤-圖片來源官網

  • 3、Broker通過主從模式來保證高可用，Broker支援Master和Slave同步複製、Master和Slave非同步複製模式，生產者的消息都是發送給Master，但是消費既可以從Master消費，也可以從Slave消費。同步複製模式可以保證即使Master宕機，消息肯定在Slave中有備份，保證了消息不會丟失。

消費

從Consumer角度分析，如何保證消息被成功消費？

  • Consumer保證消息成功消費的關鍵在於確認的時機，不要在收到消息後就立即發送消費確認，而是應該在執行完所有消費業務邏輯之後，再發送消費確認。因為消息隊列維護了消費的位置，邏輯執行失敗了，沒有確認，再去隊列拉取消息，就還是之前的一條。

10.如何處理消息重複的問題呢？

對分散式消息隊列來說，同時做到確保一定投遞和不重複投遞是很難的，就是所謂的「有且僅有一次」 。 RocketMQ擇了確保一定投遞，保證消息不丟失，但有可能造成消息重複。

處理消息重複問題，主要有業務端自己保證，主要的方式有兩種：業務冪等和消息去重。

消息重複處理

業務冪等：第一種是保證消費邏輯的冪等性，也就是多次調用和一次調用的效果是一樣的。這樣一來，不管消息消費多少次，對業務都沒有影響。

消息去重：第二種是業務端，對重複的消息就不再消費了。這種方法，需要保證每條消息都有一個惟一的編號，通常是業務相關的，比如訂單號，消費的記錄需要落庫，而且需要保證和消息確認這一步的原子性。

具體做法是可以建立一個消費記錄表，拿到這個消息做資料庫的insert操作。給這個消息做一個唯一主鍵（primary key）或者唯一約束，那麼就算出現重複消費的情況，就會導致主鍵衝突，那麼就不再處理這條消息。

11.怎麼處理消息積壓？

發生了消息積壓，這時候就得想辦法趕緊把積壓的消息消費完，就得考慮提高消費能力，一般有兩種辦法：

消息積壓處理

  • 消費者擴容：如果當前Topic的Message Queue的數量大於消費者數量，就可以對消費者進行擴容，增加消費者，來提高消費能力，儘快把積壓的消息消費玩。
  • 消息遷移Queue擴容：如果當前Topic的Message Queue的數量小於或者等於消費者數量，這種情況，再擴容消費者就沒什麼用，就得考慮擴容Message Queue。可以新建一個臨時的Topic，臨時的Topic多設置一些Message Queue，然後先用一些消費者把消費的數據丟到臨時的Topic，因為不用業務處理，只是轉發一下消息，還是很快的。接下來用擴容的消費者去消費新的Topic里的數據，消費完了之後，恢復原狀。

消息遷移擴容消費

12.順序消息如何實現？

順序消息是指消息的消費順序和產生順序相同，在有些業務邏輯下，必須保證順序，比如訂單的生成、付款、發貨，這個消息必須按順序處理才行。

順序消息

順序消息分為全局順序消息和部分順序消息，全局順序消息指某個 Topic 下的所有消息都要保證順序；

部分順序消息只要保證每一組消息被順序消費即可，比如訂單消息，只要保證同一個訂單 ID 個消息能按順序消費即可。

部分順序消息

部分順序消息相對比較好實現，生產端需要做到把同 ID 的消息發送到同一個 Message Queue ；在消費過程中，要做到從同一個Message Queue讀取的消息順序處理——消費端不能並發處理順序消息，這樣才能達到部分有序。

部分順序消息

發送端使用 MessageQueueSelector 類來控制 把消息發往哪個 Message Queue 。

順序消息生產-例子來源官方

消費端通過使用 MessageListenerOrderly 來解決單 Message Queue 的消息被並發處理的問題。
消費端述

全局順序消息

RocketMQ 默認情況下不保證順序，比如創建一個 Topic ，默認八個寫隊列，八個讀隊列，這時候一條消息可能被寫入任意一個隊列里；在數據的讀取過程中，可能有多個 Consumer ，每個 Consumer 也可能啟動多個執行緒並行處理，所以消息被哪個 Consumer 消費，被消費的順序和寫人的順序是否一致是不確定的。

要保證全局順序消息， 需要先把 Topic 的讀寫隊列數設置為 一，然後Producer Consumer 的並發設置，也要是一。簡單來說，為了保證整個 Topic全局消息有序，只能消除所有的並發處理，各部分都設置成單執行緒處理 ，這時候就完全犧牲RocketMQ的高並發、高吞吐的特性了。

全局順序消息

13.如何實現消息過濾？

有兩種方案：

  • 一種是在 Broker 端按照 Consumer 的去重邏輯進行過濾，這樣做的好處是避免了無用的消息傳輸到 Consumer 端，缺點是加重了 Broker 的負擔，實現起來相對複雜。
  • 另一種是在 Consumer 端過濾，比如按照消息設置的 tag 去重，這樣的好處是實現起來簡單，缺點是有大量無用的消息到達了 Consumer 端只能丟棄不處理。

一般採用Cosumer端過濾，如果希望提高吞吐量，可以採用Broker過濾。

對消息的過濾有三種方式：

消息過濾

  • 根據Tag過濾：這是最常見的一種，用起來高效簡單

    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_EXAMPLE");
consumer.subscribe("TOPIC", "TAGA || TAGB || TAGC");

  • SQL 表達式過濾：SQL表達式過濾更加靈活

    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
// 只有訂閱的消息有這個屬性a, a >=0 and a <= 3
consumer.subscribe("TopicTest", MessageSelector.bySql("a between 0 and 3");
consumer.registerMessageListener(new MessageListenerConcurrently() {
   @Override
   public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
       return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
   }
});
consumer.start();

  • Filter Server 方式：最靈活，也是最複雜的一種方式，允許用戶自定義函數進行過濾

14.延時消息了解嗎？

電商的訂單超時自動取消，就是一個典型的利用延時消息的例子，用戶提交了一個訂單，就可以發送一個延時消息，1h後去檢查這個訂單的狀態，如果還是未付款就取消訂單釋放庫存。

RocketMQ是支援延時消息的，只需要在生產消息的時候設置消息的延時級別：

      // 實例化一個生產者來產生延時消息
      DefaultMQProducer producer = new DefaultMQProducer("ExampleProducerGroup");
      // 啟動生產者
      producer.start();
      int totalMessagesToSend = 100;
      for (int i = 0; i < totalMessagesToSend; i++) {
          Message message = new Message("TestTopic", ("Hello scheduled message " + i).getBytes());
          // 設置延時等級3,這個消息將在10s之後發送(現在只支援固定的幾個時間,詳看delayTimeLevel)
          message.setDelayTimeLevel(3);
          // 發送消息
          producer.send(message);
      }

但是目前RocketMQ支援的延時級別是有限的：

private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

RocketMQ怎麼實現延時消息的？

簡單，八個字：臨時存儲+定時任務。

Broker收到延時消息了，會先發送到主題（SCHEDULE_TOPIC_XXXX）的相應時間段的Message Queue中，然後通過一個定時任務輪詢這些隊列，到期後，把消息投遞到目標Topic的隊列中，然後消費者就可以正常消費這些消息。

延遲消息處理流程-圖片來源見水印

15.怎麼實現分散式消息事務的？半消息？

半消息：是指暫時還不能被 Consumer 消費的消息，Producer 成功發送到 Broker 端的消息，但是此消息被標記為 「暫不可投遞」 狀態，只有等 Producer 端執行完本地事務後經過二次確認了之後，Consumer 才能消費此條消息。

依賴半消息，可以實現分散式消息事務，其中的關鍵在於二次確認以及消息回查：

RocketMQ實現消息事務

  • 1、Producer 向 broker 發送半消息
  • 2、Producer 端收到響應，消息發送成功，此時消息是半消息，標記為 「不可投遞」 狀態，Consumer 消費不了。
  • 3、Producer 端執行本地事務。
  • 4、正常情況本地事務執行完成，Producer 向 Broker 發送 Commit/Rollback，如果是 Commit，Broker 端將半消息標記為正常消息，Consumer 可以消費，如果是 Rollback，Broker 丟棄此消息。
  • 5、異常情況，Broker 端遲遲等不到二次確認。在一定時間後，會查詢所有的半消息，然後到 Producer 端查詢半消息的執行情況。
  • 6、Producer 端查詢本地事務的狀態
  • 7、根據事務的狀態提交 commit/rollback 到 broker 端。（5，6，7 是消息回查）
  • 8、消費者段消費到消息之後，執行本地事務，執行本地事務。

16.死信隊列知道嗎？

死信隊列用於處理無法被正常消費的消息，即死信消息。

當一條消息初次消費失敗，消息隊列 RocketMQ 會自動進行消息重試；達到最大重試次數後，若消費依然失敗，則表明消費者在正常情況下無法正確地消費該消息，此時，消息隊列 RocketMQ 不會立刻將消息丟棄，而是將其發送到該消費者對應的特殊隊列中，該特殊隊列稱為死信隊列。

死信消息的特點：

  • 不會再被消費者正常消費。
  • 有效期與正常消息相同，均為 3 天，3 天后會被自動刪除。因此，需要在死信消息產生後的 3 天內及時處理。

死信隊列的特點：

  • 一個死信隊列對應一個 Group ID， 而不是對應單個消費者實例。
  • 如果一個 Group ID 未產生死信消息，消息隊列 RocketMQ 不會為其創建相應的死信隊列。
  • 一個死信隊列包含了對應 Group ID 產生的所有死信消息，不論該消息屬於哪個 Topic。

RocketMQ 控制台提供對死信消息的查詢、導出和重發的功能。

17.如何保證RocketMQ的高可用？

NameServer因為是無狀態，且不相互通訊的，所以只要集群部署就可以保證高可用。

NameServer集群

RocketMQ的高可用主要是在體現在Broker的讀和寫的高可用，Broker的高可用是通過集群和主從實現的。

Broker集群、主從示意圖

Broker可以配置兩種角色：Master和Slave，Master角色的Broker支援讀和寫，Slave角色的Broker只支援讀，Master會向Slave同步消息。

也就是說Producer只能向Master角色的Broker寫入消息，Cosumer可以從Master和Slave角色的Broker讀取消息。

Consumer 的配置文件中，並不需要設置是從 Master 讀還是從 Slave讀，當 Master 不可用或者繁忙的時候， Consumer 的讀請求會被自動切換到從 Slave。有了自動切換 Consumer 這種機制，當一個 Master 角色的機器出現故障後，Consumer 仍然可以從 Slave 讀取消息，不影響 Consumer 讀取消息，這就實現了讀的高可用。

如何達到發送端寫的高可用性呢？在創建 Topic 的時候，把 Topic 的多個Message Queue 創建在多個 Broker 組上（相同 Broker 名稱，不同 brokerId機器組成 Broker 組），這樣當 Broker 組的 Master 不可用後，其他組Master 仍然可用， Producer 仍然可以發送消息 RocketMQ 目前還不支援把Slave自動轉成 Master ，如果機器資源不足，需要把 Slave 轉成 Master ，則要手動停止 Slave 色的 Broker ，更改配置文件，用新的配置文件啟動 Broker。

原理

18.說一下RocketMQ的整體工作流程？

簡單來說，RocketMQ是一個分散式消息隊列，也就是消息隊列+分散式系統。

作為消息隊列，它是發–存–收的一個模型，對應的就是Producer、Broker、Cosumer；作為分散式系統，它要有服務端、客戶端、註冊中心，對應的就是Broker、Producer/Consumer、NameServer

所以我們看一下它主要的工作流程：RocketMQ由NameServer註冊中心集群、Producer生產者集群、Consumer消費者集群和若干Broker（RocketMQ進程）組成：

  1. Broker在啟動的時候去向所有的NameServer註冊，並保持長連接，每30s發送一次心跳
  2. Producer在發送消息的時候從NameServer獲取Broker伺服器地址，根據負載均衡演算法選擇一台伺服器來發送消息
  3. Conusmer消費消息的時候同樣從NameServer獲取Broker地址，然後主動拉取消息來消費

RocketMQ整體工作流程

19.為什麼RocketMQ不使用Zookeeper作為註冊中心呢？

Kafka我們都知道採用Zookeeper作為註冊中心——當然也開始逐漸去Zookeeper，RocketMQ不使用Zookeeper其實主要可能從這幾方面來考慮：

  1. 基於可用性的考慮，根據CAP理論，同時最多只能滿足兩個點，而Zookeeper滿足的是CP，也就是說Zookeeper並不能保證服務的可用性，Zookeeper在進行選舉的時候，整個選舉的時間太長，期間整個集群都處於不可用的狀態，而這對於一個註冊中心來說肯定是不能接受的，作為服務發現來說就應該是為可用性而設計。
  2. 基於性能的考慮，NameServer本身的實現非常輕量，而且可以通過增加機器的方式水平擴展，增加集群的抗壓能力，而Zookeeper的寫是不可擴展的，Zookeeper要解決這個問題只能通過劃分領域，劃分多個Zookeeper集群來解決，首先操作起來太複雜，其次這樣還是又違反了CAP中的A的設計，導致服務之間是不連通的。
  3. 持久化的機制來帶的問題，ZooKeeper 的 ZAB 協議對每一個寫請求，會在每個 ZooKeeper 節點上保持寫一個事務日誌，同時再加上定期的將記憶體數據鏡像（Snapshot）到磁碟來保證數據的一致性和持久性，而對於一個簡單的服務發現的場景來說，這其實沒有太大的必要，這個實現方案太重了。而且本身存儲的數據應該是高度訂製化的。
  4. 消息發送應該弱依賴註冊中心，而RocketMQ的設計理念也正是基於此，生產者在第一次發送消息的時候從NameServer獲取到Broker地址後快取到本地，如果NameServer整個集群不可用，短時間內對於生產者和消費者並不會產生太大影響。

20.Broker是怎麼保存數據的呢？

RocketMQ主要的存儲文件包括CommitLog文件、ConsumeQueue文件、Indexfile文件。
消息存儲文件

消息存儲的整體的設計：

消息存儲整體設計-來源官網

  • CommitLog：消息主體以及元數據的存儲主體，存儲Producer端寫入的消息主體內容,消息內容不是定長的。單個文件大小默認1G, 文件名長度為20位，左邊補零，剩餘為起始偏移量，比如00000000000000000000代表了第一個文件，起始偏移量為0，文件大小為1G=1073741824；當第一個文件寫滿了，第二個文件為00000000001073741824，起始偏移量為1073741824，以此類推。消息主要是順序寫入日誌文件，當文件滿了，寫入下一個文件。

    CommitLog文件保存於${Rocket_Home}/store/commitlog目錄中，從圖中我們可以明顯看出來文件名的偏移量，每個文件默認1G，寫滿後自動生成一個新的文件。

    CommitLog

  • ConsumeQueue：消息消費隊列，引入的目的主要是提高消息消費的性能，由於RocketMQ是基於主題topic的訂閱模式，消息消費是針對主題進行的，如果要遍歷commitlog文件中根據topic檢索消息是非常低效的。

    Consumer即可根據ConsumeQueue來查找待消費的消息。其中，ConsumeQueue（邏輯消費隊列）作為消費消息的索引，保存了指定Topic下的隊列消息在CommitLog中的起始物理偏移量offset，消息大小size和消息Tag的HashCode值。

    ConsumeQueue文件可以看成是基於Topic的CommitLog索引文件，故ConsumeQueue文件夾的組織方式如下：topic/queue/file三層組織結構，具體存儲路徑為：$HOME/store/consumequeue/{topic}/{queueId}/{fileName}。同樣ConsumeQueue文件採取定長設計，每一個條目共20個位元組，分別為8位元組的CommitLog物理偏移量、4位元組的消息長度、8位元組tag hashcode，單個文件由30W個條目組成，可以像數組一樣隨機訪問每一個條目，每個ConsumeQueue文件大小約5.72M；

    Comsumer Queue

  • IndexFile：IndexFile（索引文件）提供了一種可以通過key或時間區間來查詢消息的方法。Index文件的存儲位置是：\(HOME \store\index\){fileName}，文件名fileName是以創建時的時間戳命名的，固定的單個IndexFile文件大小約為400M，一個IndexFile可以保存 2000W個索引，IndexFile的底層存儲設計為在文件系統中實現HashMap結構，故RocketMQ的索引文件其底層實現為hash索引。

    IndexFile文件示意圖-來源參考[2]

總結一下：RocketMQ採用的是混合型的存儲結構，即為Broker單個實例下所有的隊列共用一個日誌數據文件（即為CommitLog）來存儲。

RocketMQ的混合型存儲結構(多個Topic的消息實體內容都存儲於一個CommitLog中)針對Producer和Consumer分別採用了數據和索引部分相分離的存儲結構，Producer發送消息至Broker端，然後Broker端使用同步或者非同步的方式對消息刷盤持久化，保存至CommitLog中。

只要消息被刷盤持久化至磁碟文件CommitLog中，那麼Producer發送的消息就不會丟失。正因為如此，Consumer也就肯定有機會去消費這條消息。當無法拉取到消息後，可以等下一次消息拉取，同時服務端也支援長輪詢模式，如果一個消息拉取請求未拉取到消息，Broker允許等待30s的時間，只要這段時間內有新消息到達，將直接返回給消費端。

這裡，RocketMQ的具體做法是，使用Broker端的後台服務執行緒—ReputMessageService不停地分發請求並非同步構建ConsumeQueue（邏輯消費隊列）和IndexFile（索引文件）數據。

總結

21.說說RocketMQ怎麼對文件進行讀寫的？

RocketMQ對文件的讀寫巧妙地利用了作業系統的一些高效文件讀寫方式——PageCache、順序讀寫、零拷貝。

  • PageCache、順序讀取

在RocketMQ中，ConsumeQueue邏輯消費隊列存儲的數據較少，並且是順序讀取，在page cache機制的預讀取作用下，Consume Queue文件的讀性能幾乎接近讀記憶體，即使在有消息堆積情況下也不會影響性能。而對於CommitLog消息存儲的日誌數據文件來說，讀取消息內容時候會產生較多的隨機訪問讀取，嚴重影響性能。如果選擇合適的系統IO調度演算法，比如設置調度演算法為「Deadline」（此時塊存儲採用SSD的話），隨機讀的性能也會有所提升。

頁快取（PageCache)是OS對文件的快取，用於加速對文件的讀寫。一般來說，程式對文件進行順序讀寫的速度幾乎接近於記憶體的讀寫速度，主要原因就是由於OS使用PageCache機制對讀寫訪問操作進行了性能優化，將一部分的記憶體用作PageCache。對於數據的寫入，OS會先寫入至Cache內，隨後通過非同步的方式由pdflush內核執行緒將Cache內的數據刷盤至物理磁碟上。對於數據的讀取，如果一次讀取文件時出現未命中PageCache的情況，OS從物理磁碟上訪問讀取文件的同時，會順序對其他相鄰塊的數據文件進行預讀取。

  • 零拷貝

另外，RocketMQ主要通過MappedByteBuffer對文件進行讀寫操作。其中，利用了NIO中的FileChannel模型將磁碟上的物理文件直接映射到用戶態的記憶體地址中（這種Mmap的方式減少了傳統IO，將磁碟文件數據在作業系統內核地址空間的緩衝區，和用戶應用程式地址空間的緩衝區之間來回進行拷貝的性能開銷），將對文件的操作轉化為直接對記憶體地址進行操作，從而極大地提高了文件的讀寫效率（正因為需要使用記憶體映射機制，故RocketMQ的文件存儲都使用定長結構來存儲，方便一次將整個文件映射至記憶體）。

說說什麼是零拷貝?

在作業系統中，使用傳統的方式，數據需要經歷幾次拷貝，還要經歷用戶態/內核態切換。

傳統文件傳輸示意圖-來源《圖解作業系統》

  1. 從磁碟複製數據到內核態記憶體；
  2. 從內核態記憶體複製到用戶態記憶體；
  3. 然後從用戶態記憶體複製到網路驅動的內核態記憶體；
  4. 最後是從網路驅動的內核態記憶體複製到網卡中進行傳輸。

所以，可以通過零拷貝的方式，減少用戶態與內核態的上下文切換和記憶體拷貝的次數，用來提升I/O的性能。零拷貝比較常見的實現方式是mmap，這種機制在Java中是通過MappedByteBuffer實現的。

mmap示意圖-來源《圖解作業系統》

22.消息刷盤怎麼實現的呢？

RocketMQ提供了兩種刷盤策略：同步刷盤和非同步刷盤

  • 同步刷盤：在消息達到Broker的記憶體之後，必須刷到commitLog日誌文件中才算成功，然後返回Producer數據已經發送成功。
  • 非同步刷盤：非同步刷盤是指消息達到Broker記憶體後就返回Producer數據已經發送成功，會喚醒一個執行緒去將數據持久化到CommitLog日誌文件中。

Broker 在消息的存取時直接操作的是記憶體（記憶體映射文件），這可以提供系統的吞吐量，但是無法避免機器掉電時數據丟失，所以需要持久化到磁碟中。

刷盤的最終實現都是使用NIO中的 MappedByteBuffer.force() 將映射區的數據寫入到磁碟，如果是同步刷盤的話，在Broker把消息寫到CommitLog映射區後，就會等待寫入完成。

非同步而言，只是喚醒對應的執行緒，不保證執行的時機，流程如圖所示。

非同步刷盤

22.能說下 RocketMQ 的負載均衡是如何實現的？

RocketMQ中的負載均衡都在Client端完成，具體來說的話，主要可以分為Producer端發送消息時候的負載均衡和Consumer端訂閱消息的負載均衡。

Producer的負載均衡

Producer端在發送消息的時候，會先根據Topic找到指定的TopicPublishInfo，在獲取了TopicPublishInfo路由資訊後，RocketMQ的客戶端在默認方式下selectOneMessageQueue()方法會從TopicPublishInfo中的messageQueueList中選擇一個隊列（MessageQueue）進行發送消息。具這裡有一個sendLatencyFaultEnable開關變數，如果開啟，在隨機遞增取模的基礎上，再過濾掉not available的Broker代理。

[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-VuMGUv6B-1649246218316)(//cdn.jsdelivr.net/gh/fighter3/picgo/img/20220405230011.png)]

所謂的”latencyFaultTolerance”，是指對之前失敗的，按一定的時間做退避。例如，如果上次請求的latency超過550Lms，就退避3000Lms；超過1000L，就退避60000L；如果關閉，採用隨機遞增取模的方式選擇一個隊列（MessageQueue）來發送消息，latencyFaultTolerance機制是實現消息發送高可用的核心關鍵所在。

Consumer的負載均衡

在RocketMQ中，Consumer端的兩種消費模式（Push/Pull）都是基於拉模式來獲取消息的，而在Push模式只是對pull模式的一種封裝，其本質實現為消息拉取執行緒在從伺服器拉取到一批消息後，然後提交到消息消費執行緒池後，又「馬不停蹄」的繼續向伺服器再次嘗試拉取消息。如果未拉取到消息，則延遲一下又繼續拉取。在兩種基於拉模式的消費方式（Push/Pull）中，均需要Consumer端知道從Broker端的哪一個消息隊列中去獲取消息。因此，有必要在Consumer端來做負載均衡，即Broker端中多個MessageQueue分配給同一個ConsumerGroup中的哪些Consumer消費。

  1. Consumer端的心跳包發送

在Consumer啟動後，它就會通過定時任務不斷地向RocketMQ集群中的所有Broker實例發送心跳包（其中包含了，消息消費分組名稱、訂閱關係集合、消息通訊模式和客戶端id的值等資訊）。Broker端在收到Consumer的心跳消息後，會將它維護在ConsumerManager的本地快取變數—consumerTable，同時並將封裝後的客戶端網路通道資訊保存在本地快取變數—channelInfoTable中，為之後做Consumer端的負載均衡提供可以依據的元數據資訊。

  1. Consumer端實現負載均衡的核心類—RebalanceImpl

    在Consumer實例的啟動流程中的啟動MQClientInstance實例部分，會完成負載均衡服務執行緒—RebalanceService的啟動（每隔20s執行一次）。

    通過查看源碼可以發現，RebalanceService執行緒的run()方法最終調用的是RebalanceImpl類的rebalanceByTopic()方法，這個方法是實現Consumer端負載均衡的核心。

    rebalanceByTopic()方法會根據消費者通訊類型為「廣播模式」還是「集群模式」做不同的邏輯處理。這裡主要來看下集群模式下的主要處理流程：

img

(1) 從rebalanceImpl實例的本地快取變數—topicSubscribeInfoTable中，獲取該Topic主題下的消息消費隊列集合（mqSet）；

(2) 根據topic和consumerGroup為參數調用mQClientFactory.findConsumerIdList()方法向Broker端發送通訊請求，獲取該消費組下消費者Id列表；

(3) 先對Topic下的消息消費隊列、消費者Id排序，然後用消息隊列分配策略演算法（默認為：消息隊列的平均分配演算法），計算出待拉取的消息隊列。這裡的平均分配演算法，類似於分頁的演算法，將所有MessageQueue排好序類似於記錄，將所有消費端Consumer排好序類似頁數，並求出每一頁需要包含的平均size和每個頁面記錄的範圍range，最後遍歷整個range而計算出當前Consumer端應該分配到的的MessageQueue。

Cosumer分配

(4) 然後，調用updateProcessQueueTableInRebalance()方法，具體的做法是，先將分配到的消息隊列集合（mqSet）與processQueueTable做一個過濾比對。
負載均衡示意圖-來源官網

  • 上圖中processQueueTable標註的紅色部分，表示與分配到的消息隊列集合mqSet互不包含。將這些隊列設置Dropped屬性為true，然後查看這些隊列是否可以移除出processQueueTable快取變數，這裡具體執行removeUnnecessaryMessageQueue()方法，即每隔1s 查看是否可以獲取當前消費處理隊列的鎖，拿到的話返回true。如果等待1s後，仍然拿不到當前消費處理隊列的鎖則返回false。如果返回true，則從processQueueTable快取變數中移除對應的Entry；
  • 上圖中processQueueTable的綠色部分，表示與分配到的消息隊列集合mqSet的交集。判斷該ProcessQueue是否已經過期了，在Pull模式的不用管，如果是Push模式的，設置Dropped屬性為true，並且調用removeUnnecessaryMessageQueue()方法，像上面一樣嘗試移除Entry；
  • 最後，為過濾後的消息隊列集合（mqSet）中的每個MessageQueue創建一個ProcessQueue對象並存入RebalanceImpl的processQueueTable隊列中（其中調用RebalanceImpl實例的computePullFromWhere(MessageQueue mq)方法獲取該MessageQueue對象的下一個進度消費值offset，隨後填充至接下來要創建的pullRequest對象屬性中），並創建拉取請求對象—pullRequest添加到拉取列表—pullRequestList中，最後執行dispatchPullRequest()方法，將Pull消息的請求對象PullRequest依次放入PullMessageService服務執行緒的阻塞隊列pullRequestQueue中，待該服務執行緒取出後向Broker端發起Pull消息的請求。其中，可以重點對比下，RebalancePushImpl和RebalancePullImpl兩個實現類的dispatchPullRequest()方法不同，RebalancePullImpl類裡面的該方法為空。

消息消費隊列在同一消費組不同消費者之間的負載均衡，其核心設計理念是在一個消息消費隊列在同一時間只允許被同一消費組內的一個消費者消費，一個消息消費者能同時消費多個消息隊列。

23.RocketMQ消息長輪詢了解嗎？

所謂的長輪詢，就是Consumer 拉取消息，如果對應的 Queue 如果沒有數據，Broker 不會立即返回，而是把 PullReuqest hold起來，等待 queue 有了消息後，或者長輪詢阻塞時間到了，再重新處理該 queue 上的所有 PullRequest。

長輪詢簡單示意圖

  • PullMessageProcessor#processRequest

                    //如果沒有拉到數據
                case ResponseCode.PULL_NOT_FOUND:
                    // broker 和 consumer 都允許 suspend，默認開啟
                    if (brokerAllowSuspend && hasSuspendFlag) {
                        long pollingTimeMills = suspendTimeoutMillisLong;
                        if (!this.brokerController.getBrokerConfig().isLongPollingEnable()) {
                            pollingTimeMills = this.brokerController.getBrokerConfig().getShortPollingTimeMills();
                        }

                        String topic = requestHeader.getTopic();
                        long offset = requestHeader.getQueueOffset();
                        int queueId = requestHeader.getQueueId();
                        //封裝一個PullRequest
                        PullRequest pullRequest = new PullRequest(request, channel, pollingTimeMills,
                                this.brokerController.getMessageStore().now(), offset, subscriptionData, messageFilter);
                        //把PullRequest掛起來
                        this.brokerController.getPullRequestHoldService().suspendPullRequest(topic, queueId, pullRequest);
                        response = null;
                        break;
                    }

掛起的請求，有一個服務執行緒會不停地檢查，看queue中是否有數據，或者超時。

  • PullRequestHoldService#run()
    @Override
    public void run() {
        log.info("{} service started", this.getServiceName());
        while (!this.isStopped()) {
            try {
                if (this.brokerController.getBrokerConfig().isLongPollingEnable()) {
                    this.waitForRunning(5 * 1000);
                } else {
                    this.waitForRunning(this.brokerController.getBrokerConfig().getShortPollingTimeMills());
                }

                long beginLockTimestamp = this.systemClock.now();
                //檢查hold住的請求
                this.checkHoldRequest();
                long costTime = this.systemClock.now() - beginLockTimestamp;
                if (costTime > 5 * 1000) {
                    log.info("[NOTIFYME] check hold request cost {} ms.", costTime);
                }
            } catch (Throwable e) {
                log.warn(this.getServiceName() + " service has exception. ", e);
            }
        }

        log.info("{} service end", this.getServiceName());
    }

參考：

[1]. 《RocketMQ實戰與原理解析》

[2]. 《RocketMQ技術內幕》

[3]. 面試被問到RocketMq，我懵了

[4]. 艾小仙《我要進大廠》

[5]. //dreamcat.ink/java-interview/docs/knows/classify/dis/RocketMQ/

[6]. 《淺入淺出》-RocketMQ

[7].十二張圖，踹開消息隊列的大門

[8].mq的那些破事兒，你不好奇嗎?

[9]. 消息冪等（去重）如何解決？來看看這個方案！

[10] .七萬字，151張圖，通宵整理消息隊列核心知識點總結！這次徹底掌握MQ！

[11].極客時間 《消息隊列高手課》

[12].RocketMQ官網

✨面渣逆襲系列：

在這裡插入圖片描述

Tags:

VirMach 便宜 VPS

QNews

QNews