一張圖進階 RocketMQ - 消息發送

前 言

三此君看了好幾本書，看了很多遍源碼整理的 一張圖進階 RocketMQ 圖片鏈接，關於 RocketMQ 你只需要記住這張圖！覺得不錯的話，記得點贊關注哦。

一张图进阶 RocketMQ 图片链接

【重要】影片在 B 站同步更新，歡迎圍觀，輕輕鬆鬆漲姿勢。一張圖進階 RocketMQ-消息發送（影片版）

本文是「一張圖進階 RocketMQ」 系列第 3 篇，對 RocketMQ 不了解的同學可以先看看三此君的
一張圖進階 RocketMQ-整體架構，一張圖進階 RocketMQ - NameServer。
在了解了 RocketMQ 的整體架構之後，我們來深入的分析下生產者消息發送的設計與實現。本文從一個生產者示例開始，以兩行程式碼為切入點，逐步剖析生產者啟動流程以及同步消息發送流程。

生產者示例

消息發送分為同步消息、非同步消息和單向消息，簡單來說：

  • 同步消息：消息發送之後會等待 Broker 響應，並把響應結果傳遞給業務執行緒，整個過程業務執行緒在等待。
  • 非同步消息：調用非同步發送 API，Producer 把消息發送請求放進執行緒池就返回。邏輯處理，網路請求都在執行緒池中進行，等結果處理完之後回調業務定義好的回調函數。
  • 單向消息：只負責發送消息，不管發送結果。

我們先來回顧下同步消息發送的例子：

public class SyncProducer {
    public static void main(String[] args) throws Exception {
        // 實例化消息生產者Producer
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
        // 設置NameServer的地址
        producer.setNamesrvAddr("localhost:9876");
        // 啟動Producer實例
        producer.start();
        // 創建消息，並指定Topic，Tag和消息體
        Message msg = new Message("sancijun","order", "orderId", "我一定會關注三此君".getBytes("UTF-8")); 
        // 發送消息到一個Broker
        SendResult sendResult = producer.send(msg);
      	// 通過sendResult返回消息是否成功送達
        System.out.printf("%s%n", sendResult);
        // 如果不再發送消息，關閉Producer實例。
        producer.shutdown();
    }
}
  • 首先，實例化一個生產者 producer，並告訴它 NameServer 的地址，這樣生產者才能從 NameServer 獲取路由資訊。
  • 然後 producer 得做一些初始化（這是很關鍵的步驟），它要和 NameServer 通訊，要先初始化通訊模組等。
  • producer 已經準備好了，那得準備好要發的內容，把 "我一定會關注三此君" 發送到 Topic=」sanicjun「。
  • 內容準備好，那 producer 就可以把消息發送出去了。producer 怎麼知道 Broker 地址呢？他會去 NameServer 獲取路由資訊，得到 Broker 的地址是 localhost:10909，然後通過網路通訊將消息發送給 Broker。
  • 生產者發送的消息通過網路傳輸給 Broker，Broker 需要對消息按照一定的結構進行存儲。存儲完成之後，把存儲結果告知生產者。

其中有兩個關鍵的地方：producer.start() 及 producer.send()，也就是生產者初始化及消息發送。我們以這兩行程式碼為切入點，看看 RocketMQ Producer 的設計與實現。

Tips：因為本文是RocketMQ 設計與實現分析，雖然不會粘貼任何源碼，但是圖文中會有大量的類名和方法名，看的時候不必執著於這些陌生的類名和方法名，三此君會解釋這些類和方法的用途。

目標：將消息發送給 Broker 進行存儲
關鍵點 1: 怎樣根據 topic+路由資訊 建立網路通道，進行消息的發送
關鍵點 2: 消息在發送過程中又經過了哪些處理？

生產者啟動

我們實例化一個生產者 DefaultMQProducer，並調用 DefaultMQProducer.start() 方法進行初始化：

生产者启动流程

啟動流程比較長，其實最重要的就是初始化了通訊模組，並啟動了多個定時任務，這些在後面的消息發送過程中都會用到：

  • 檢查配置是否合法：生產者組名是否為空、是否滿足命名規則、長度是否滿足等。

  • 啟動通訊模組服務 Netty RemotingClient：RemotingClient 是一個介面，底層使用的通訊框架是Netty，提供了實現類 NettyRemotingClient，RemotingClient 在初始化的時候實例化 Bootstrap，方便後續用來創建 SocketChannel；後文會介紹 RocketMQ 的通訊機制，大家稍安勿躁。

  • 啟動 5 個後台定時任務：定時更新 NameServerAddr 資訊，定時更新 topic 的路由資訊，定時向 Broker 發送心跳及清理下線的 Broker，定時持久化 Consumer 的 Offset 資訊，定時調整執行緒池；

    生產者每 30s 會從某台 NameServer 獲取 Topic 和 Broker 的映射關係（路由資訊）存在本地記憶體中，如果發現新的 Broker 就會和其建立長連接，每 30s 會發送心跳至 Broker 維護連接。

Tips：生產者為什麼要啟動消息拉取服務？重平衡服務是什麼？簡單來說，這兩個服務都是用於消費者的，這裡我們暫且不理會。消息拉取服務 pullMessageService 是從 Broker 拉取消息的服務 ，重平衡服務 rebalanceService 用於消費者的負載均衡，負責分配消費者可消費的消息隊列。

同步發送

總體上講，消息發送可以劃分為三個層級：

  • 業務層：準備需要發送的消息。
  • 消息處理層：獲取業務發送的 Message，經過一系列的參數檢查、消息發送準備、參數包裝等操作。
  • 通訊層：基於 Netty 封裝的一個網路通訊服務，將消息發送給 Broker。

同步消息发送

我們通過前面的示例來看整個同步消息發送的處理流程，整個過程我們的主要目標就是把消息發送到 Broker：

  • 第一步：業務層構建待發送消息 Message msg = new Message("sancijun","order", "orderId", "我一定會關注三此君".getBytes("UTF-8"));

  • 第二步：然後我們調用 producer.send(msg) 發送消息，可是 producer 怎麼知道發給誰呢？消息本身又需要經過哪些處理呢？我們進入調用鏈直到 sendDefaultImpl

    • 檢查消息是否為空，消息的 Topic 的名字是否為空或者是否符合規範，消息體大小是否符合要求，最大值為4MB，可以通過 maxMessageSize 進行設置。

    • 執行 tryToFindTopicPublishInfo() 方法：獲取 Topic 路由資訊，如果不存在則拋出異常。如果本地快取沒有路由資訊，就通過Namesrv 獲取路由資訊，更新到本地。消息構建的時候我們指定了消息所屬 Topic，根據 Topic 路由資訊我們可以找到對應的 Broker。

      Tips：從 NameServer 獲取的路由資訊 TopicRouteData 會包含指定 Topic 的 topicQueueTable、brokerAddrTable。在 NameServer 集群元數據管理部分我們講過，通過 topicName 從 topicQueueTable 獲取對應的 brokerName，再根據 brokerName 從 brokerAddrTable 中獲取 Broker IP 地址。

    • 計算消息發送的重試次數，同步重試和非同步重試的執行方式是不同的。在同步發送情況下如果發送失敗會默認重投兩次（默認retryTimesWhenSendFailed = 2），並且不會選擇上次失敗的 Broker，會向其他 Broker 投遞。

    • 執行隊列選擇方法 selectOneMessageQueue()。根據 lastBrokerName（上次發送消息失敗的 Broker 的名字）和 Topic 路由資訊選一個 MessageQueue。
      首次發送時 lastBrokerName 為 null，採用輪詢策略選擇一個 MessageQueue。如果上次發送失敗，也是採用輪詢策略選擇一個 MessageQueue，但是會跳過上次發送失敗 Broker 的 MessageQueue，也就是換一個 Broker 發送。

      Tips：選擇一個 MessageQueue，什麼是 MessageQueue 呢？這和 Broker 的存儲結構相關，我們會在存儲部分詳細介紹，這裡先說結論，我們創建 Topic 時指定了這個 Topic 的讀寫隊列數，每個 MessageQueue 有不同的 queueId(0-3)。

      我們也可以通過sendLatencyFaultEnable 來設置是否總是發送到延遲級別較低的Broker，默認值為False，我么這裡就不展開討論了。

    • 執行 sendKernelImpl() 方法。

  • 第三步：sendDefaultImpl 做了一系列邏輯處理，我們已經得到了待發送的 BrokerName，而我們的目標是把消息發送到 Broker。sendKernelImpl 方法是發送消息的核心方法，主要用於準備通訊層的入參（比如Broker地址、請求體等），將請求傳遞給通訊層。

    • 根據 MessageQueue.brokerName 獲取 Broker IP 地址，給 message 添加全局唯一 ID。

      Tips：sendKernelImpl 也有很多的邏輯處理，我們暫時先略過這裡的壓縮、事務消息、鉤子函數、重試消息：

      對大於4k的普通消息進行壓縮，並設置消息的系統標記為MessageSysFlag.COMPRESSED_FLAG。

      如果是事務Prepared消息，則設置消息的系統標記為MessageSysFlag.TRANSACTION_PREPARED_TYPE

      如果註冊了消息發送鉤子函數，則執行消息發送之前的增強邏輯，通過DefaultMQProducerImpl#registerSendMessageHook註冊鉤子處理類，並且可以註冊多個。

      構建發送消息請求頭：生產者組、主題名稱、默認創建主題Key、該主題在單個Broker默認隊列數、隊列ID（隊列序號）、消息系統標記（MessageSysFlag）、消息發送時間、消息標記、消息擴展屬性、消息重試次數、是否是批量消息等

      處理重試消息。

    • 調用 MQClientAPIImpl.sendMessage()，首先構建一個遠程請求 RemotingCommand，根據發送類型（同步或非同步）調用不同的通訊層實現方法。我們這裡是同步消息，則調用 RemotingClient.invokeSync()。

    • 處理返回結果，將通訊層返回的結果封裝成 SendResult 對象返回給業務層。

  • 第四步：RemotingClient 是基於 Netty 實現的，熟悉 Netty 的同學已經大概知道後面的流程，不熟悉的同學也沒有關係，這裡先混個眼熟，下面我們會對 Netty 做簡單的介紹。

    • RemotingClient.invokeSync() 先是通過 Broker Addr 獲取或者創建 Netty Channel。先從 channelTables Map 本地快取中，以Broker Addr 為 key 獲取 Channel，沒有獲取到則通過 Netty Bootstrap.connect( Broker Addr) 創建 Channel，並放入快取。
    • 然後生成<opaque, ResponseFuture>的鍵值對放入 responseTable 快取中，結果返回的時候根據 opaque 從快取中獲取結果。
    • 調用 channel.writeAndFlush() 將消息通過網路傳輸給指定 Broker。這裡是 Netty 框架的 API，已經不在 RocketMQ 範疇。
    • 調用 ResponseFuture.waitResponse() 方法，直到 Netty 接收 Broker的返回結果。其實就是執行 countDownLatch.await()。

  • 第五步：結果處理及返回。

    • Broker 處理結果返回，Netty 產生可讀事件，由 Channelhandler 處理可讀事件，這裡是 NettyClientHandler.channelRead0()接收寫入數據，處理可讀事件。
    • 然後處理返回結果，從 responseTable 取出 ResponseFuture，並執行 responseFuture.putResponse()。實際上就只執行 countDownLatch.countDown() 喚醒第四步中等待的調用執行緒，返回 Broker 的處理結果 RemotingCommand。
    • 結果層層返回，直到 MQClientAPIImpl.sendMessageSync() 出手了，這裡調用 MQClientAPIImpl.processSendResponse() 處理返回結果，封裝成 SendResult 對象返回給業務層。

到這裡，生產者已經將消息發送到指定的 Broker 了，其中包括了消息的層層校驗及封裝；還有很重要的是如何選擇一個 MessageQueue 進行發送（重試），重試是保證消息發送可靠的關鍵步驟；最後通過 Netty 將請求發送給 Broker。我們先不管 Broker 收到請求如何處理，但是要明白消息如何送到 Broker 進行存儲，需要對 Netty 有簡單的理解。

總結

以上就是 RocketMQ 消息發送的主要內容，我們簡單的總結下：

  • 生產者啟動：主要是調用 NettyRemotingClient.start() 初始化 Netty 客戶端，並啟動 5 個後台執行緒；
  • 消息發送：業務層封裝發送的消息，邏輯層進行層層校驗及封裝，輪詢策略選擇一個 MessageQueue 發送(重試)，通訊層基於 Netty 將消息發送給 Broker。

banner

參考文獻

  • RocketMQ 官方文檔

  • RocketMQ 源碼

  • 丁威, 周繼鋒. RocketMQ技術內幕：RocketMQ架構設計與實現原理. 機械工業出版社, 2019-01.

  • 李偉. RocketMQ分散式消息中間件：核心原理與最佳實踐. 電子工業出版社, 2020-08.

  • 楊開元. RocketMQ實戰與原理解析. 機械工業出版社, 2018-06.

Tags:

VirMach 便宜 VPS