天天看點
傳回

RocketMQ架構原了解析（四）：消息生産端（Producer）

RocketMQ架構原了解析（一）：整體架構

RocketMQ架構原了解析（二）：消息存儲（CommitLog）

RocketMQ架構原了解析（三）：消息索引（ConsumeQueue & IndexFile）

RocketMQ架構原了解析（四）：消息生産端（Producer）

一、概述

如果你曾經使用過RocketMQ，那麼一定對以下發送消息的代碼不陌生

DefaultMQProducer producer = new DefaultMQProducer("producerGroup");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();
Message message = new Message(topic, new byte[] {'hello, world'});
producer.send(message);

寥寥幾行代碼，便是本文要論述的全部。阿裡有句土話，叫“把複雜留給自己，把簡單交給别人”用在這裡可能最合适不過了，這5行代碼中，最重要的是

producer.start()

及

producer.send()

，也就是producer啟動及消息發送

二、Producer啟動

對應代碼

producer.start()
RocketMQ架構原了解析（四）：消息生産端（Producer）

其實僅僅一行代碼，在produer端的背景啟動了多個線程來協同工作，接下來我們逐一闡述

2.1、Netty

我們都知道，RocketMQ是一個叢集部署、跨網絡的産品，除了producer、consumer需要網絡傳輸外，資料還需要在叢集中流轉。是以一個高效、可靠的網絡元件是必不可少的。而RocketMQ選擇了netty

使用netty首先需要考慮的便是業務上的資料粘包問題，netty提供了一些較為常用的解決方案，如：固定長度（比如每次發送的消息長度均為1024byte）、固定分隔符（比如每次發送的消息長度均為1024byte）等。而RocketMQ使用的則是最為通用的head、body分離方式，即head存儲消息的長度，body存儲真正的消息資料，具體實作可參見類

o.a.r.r.n.NettyRemotingClient

而消息收發這塊，RocketMQ将所有的消息都收斂到同一個協定類

o.a.r.r.p.RemotingCommand

中，即消息發送、接收都會将其封裝在該類中，這樣做的好處是不言而喻的，即統一規範，減輕網絡協定适配不同的消息類型帶來的負擔

其中較為重要的2個 ChannelHanlder 如下

  • org.apache.rocketmq.remoting.netty.NettyEncoder
    • 消息編碼，向 broker 或 nameServer 發送消息時使用，将

      RemotingCommand

       轉換為

      byte[]

       形式

  • org.apache.rocketmq.remoting.netty.NettyDecoder
    • 消息解碼，将

      byte[]

      RemotingCommand

       對象，接收 broker 傳回的消息時，進行解碼操作

2.2、消息格式

消息格式是什麼概念？在《消息存儲》章節不是已經闡述過消息格式了嗎？其實這是兩個概念，《消息存儲》章節是消息真正落盤時候的存儲格式，本小節的消息格式是指消息以什麼樣的形态交給netty進而在網絡上進行傳輸

消息格式由MsgHeader及MsgBody組成，而消息的長度、标記、版本等重要參數都放在 header 中，body 中僅僅存儲資料，沒有額外字段；我們主要看一下 header 的資料格式

RocketMQ架構原了解析（四）：消息生産端（Producer）

而站在 netty 視角來看，不論是 msgHeader 還是 msgBody，都屬于 netty 網絡消息的body部分，是以我們可以簡單畫一張 netty 視角的消息格式

RocketMQ架構原了解析（四）：消息生産端（Producer）

2.2.1、Msg Header的自動适配

上文得知，RocketMQ将所有的消息類型、收發都收斂到類

RemotingCommand

中，但RocketMQ消息類型衆多，除了常見的消息發送、接收外，還有通過msgID查詢消息、msgKey查詢消息、擷取broker配置、清理不再使用的topic等等，用一個類适配如此多的類型，具體是如何實作的呢？當新增、修改一種類型又該怎麼應對呢？

翻看源碼便發現，

RemotingCommand

的消息頭定義為一個接口

org.apache.rocketmq.remoting.CommandCustomHeader

，不同類型的請求都實作這個接口，并在自己的子類中定義成員變量；那

RemotingCommand

的消息頭又是如何自動解析呢？

public void makeCustomHeaderToNet() {
    if (this.customHeader != null) {
        Field[] fields = getClazzFields(customHeader.getClass());
        if (null == this.extFields) {
            this.extFields = new HashMap<String, String>();
        }
        for (Field field : fields) {
            if (!Modifier.isStatic(field.getModifiers())) {
                String name = field.getName();
                if (!name.startsWith("this")) {
                    Object value = null;
                    try {
                        field.setAccessible(true);
                        value = field.get(this.customHeader);
                    } catch (Exception e) {
                        log.error("Failed to access field [{}]", name, e);
                    }

                    if (value != null) {
                        this.extFields.put(name, value.toString());
                    }
                }
            }
        }
    }
}

答案就是反射，通過反射擷取子類的全部成員屬性，并放入變量

extFields

中，

makeCustomHeaderToNet()

通過犧牲少量性能的方式，換取了程式極大的靈活性與擴充性，當新增請求類型時，僅需要編寫新請求的encode、decode，不用修改其他類型請求的代碼

RocketMQ架構原了解析（四）：消息生産端（Producer）

2.3、Topic路由資訊

2.3.1、Topic建立

發送消息的前置是需要建立一個topic，建立topic的admin指令如下

updateTopic -b <> -t <> -r <> -w <> -p <> -o <> -u <> -s <>

例如：
updateTopic -b 127.0.0.1:10911 -t testTopic -r 8 -w 8 -p 6 -o false -u false -s false

簡單介紹下每個參數的作用

  • -b

     broker 位址，表示 topic 所在 Broker，隻支援單台Broker，位址為ip:port

  • -c

     cluster 位址，表示 topic 所在 cluster，會向 cluster 中所有的 broker 發送請求

  • -t

     topic 名稱

  • -r

     可讀隊列數（預設為 8，後文還會展開）

  • -w

     可寫隊列數（預設為 8，後文還會展開）

  • -p

     指定新topic的讀寫權限 （W=2|R=4|WR=6）2表示目前topic僅可寫入資料，4表示僅可讀，6表示可讀可寫

  • -o

     set topic's order(true|false)

  • -u

     is unit topic (true|false)

  • -s

     has unit sub (true|false)

如果執行指令

updateTopic -b 127.0.0.1:8899 -t testTopic -r 8 -w 8

意味着會在127.0.0.1:8899對應的broker下建立一個topic，這個topic的讀寫隊列都是 8

那如果是這樣的場景呢：叢集A有3個master節點，當執行指令

updateTopic -c clusterName -t testTopic -r 8 -w 8

後，站在叢集A角度來看，目前topic總共建立了多少個寫隊列？其實 RocketMQ 接到這條指令後，會向3個 broker 分别發送建立 topic 的指令，這樣每個broker上都會有8個讀隊列，8個寫隊列，是以站在叢集的視角，這個 topic 總共會有 24 個讀隊列，24 個寫隊列

  • 建立流程
    • 這裡簡單提一下建立流程，建立順序 admin -> broker -> nameServer（多）
    • 因為nameServer為輕量級設計，即節點之間不會互相通信，是以建立 topic 的消息需要廣播給所有nameServer

2.3.2、writeQueueNum VS readQueueNum

首選需要明确的是，讀、寫隊列，這兩個概念是 RocketMQ 獨有的，而 kafka 中隻有一個partition的概念，不區分讀寫。一般情況下，這兩個值建議設定為相等；我們分别看一下 client 端對它們的處理 （均在類

MQClientInstance.java

中）

producer端

for (int i = 0; i < qd.getWriteQueueNums(); i++) {
    MessageQueue mq = new MessageQueue(topic, qd.getBrokerName(), i);
    info.getMessageQueueList().add(mq);
}

consumer端

for (int i = 0; i < qd.getReadQueueNums(); i++) {
    MessageQueue mq = new MessageQueue(topic, qd.getBrokerName(), i);
    mqList.add(mq);
}

如果2個隊列設定不相等，例如我們設定6個寫隊列，4個讀隊列的話：

RocketMQ架構原了解析（四）：消息生産端（Producer）

這樣，4、5号隊列中的資料一定不會被消費。

  • writeQueueNum > readQueueNum
    • 大于 readQueueNum 部分的隊列永遠不會被消費

  • writeQueueNum < readQueueNum
    • 所有隊列中的資料都會被消費，但部分讀隊列資料一直是空的

這樣設計有什麼好處呢？其實是更精細的控制了讀寫操作，例如當我們要遷移 broker 時，可以首先将寫入隊列設定為0，将用戶端引流至其他 broker 節點，等讀隊列資料也處理完畢後，再關閉 read 操作

2.3.3、路由資料格式

topic的路由資料如何由Admin發起建立，再被各個broker響應，繼而被nameServer統一組織建立的流程我們暫且不讨論，為防止發散，我們直接從producer從nameServer擷取路由資料開始。從nameServer擷取到的路由資料格式如下

public class TopicRouteData extends RemotingSerializable {
    private String orderTopicConf;
    private List<QueueData> queueDatas;
    private List<BrokerData> brokerDatas;
    private HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
}

而存放路由資料的結構是

queueDatas

brokerDatas

 
public class QueueData implements Comparable<QueueData> {
    private String brokerName;
    private int readQueueNums;
    private int writeQueueNums;
}

public class BrokerData implements Comparable<BrokerData> {
    private String cluster;
    private String brokerName;
    private HashMap<Long/* brokerId */, String/* broker address */> brokerAddrs;
}

在此，簡單闡述一下RocketMQ的cluster、brokerName、brokerId的概念

RocketMQ架構原了解析（四）：消息生産端（Producer）

上圖描述了一個cluster下有3個broker，每個broker又有1個master，2個slave組成；這也就是為什麼類

BrokerData

中有

HashMap<Long, String> brokerAddrs

變量的原因，因為可能同一個brokerName下由多個節點組成。注：master節點的編号始終為0

2.3.4、Topic路由資訊何時發生變化

這些路由資訊什麼時候發生變化呢？我們舉例說明

舉例1：某叢集有3台 master，分别向其中的2台發送了建立topic的指令，此時所有的clent端都知道這個topic的資料在這兩個broker上；這個時候通過admin向第3台broker發送建立topic指令，nameServer的路由資訊便發生了變更，等client端30秒輪訓後，便可以更新到最新的topic路由資訊

舉例2：某叢集有3台 master，topic分别在3台broker上都建立了，此時某台broker當機，nameServer将其摘除，等待30秒輪詢後，client拿到最新路由資訊

思考：client 端路由資訊的變化是依托于30秒的輪詢，如果路由資訊已經發生變化，且輪詢未發生，client端拿着舊的topic路由資訊通路叢集，一定會有短暫報錯，此處也是待優化的點

2.3.5、定時更新Topic路由資訊

RocketMQ會每隔30秒更新topic的路由資訊

RocketMQ架構原了解析（四）：消息生産端（Producer）

此處簡單留意一下

TopicRouteData

TopicPublishInfo

，其實

TopicPublishInfo

是由

TopicRouteData

變種而來，多了一個

messageQueueList

的屬性，在producer端，該屬性為寫入隊列，即某個topic所有的可寫入的隊列集合

此處抛出一個問題，如果producer隻想某個topic發送了一條消息，後續再沒有發送過，這種設計會帶來哪些問題？如果這種場景頻繁發生呢？

2.4、與Broker心跳

主要分為兩部分：

  • 1、清空無效broker
  • 2、向有效的broker發送心跳

2.4.1、清空無效的broker

由上節得知，RocketMQ會擷取所有已經注冊的topic所在的broker資訊，并将這些資訊存儲在變量

brokerAddrTable

brokerAddrTable

的存儲結構如下

ConcurrentMap<String, HashMap<Long, String>> brokerAddrTable ;
  • key: brokerName，例如一個master帶2個slave都屬于同一個brokerName
  • val:

    HashMap<Long, String>

     ，key為brokerId（其中master的brokerId固定為0），val為ip位址

如何判斷某個broker有效無效呢？判斷依據便是

MQClientInstance#topicRouteTable

，這個變量是上節中從nameServer中同步過來的，如果

brokerAddrTable

中有broker A、B、C，而

topicRouteTable

隻有A、B的話，那麼就需要從

brokerAddrTable

中删除C。

需要注意的是，在整個check及替換過程中都添加了獨占鎖

lockNamesrv

，而上節中維護更新topic路由資訊也是指定的該鎖

2.4.2、發送心跳資料

RocketMQ架構原了解析（四）：消息生産端（Producer）

此處目的僅為與broker保持網絡心跳，如果連接配接失敗或發生異常，僅會列印日志，并不會有額外操作

三、消息發送

RocketMQ架構原了解析（四）：消息生産端（Producer）

消息發送比較重要的是2點内容

  • 發送資料的負載均衡問題；RocketMQ預設采用的是輪訓的方式
  • 消息發送的方式；分同步、異步、單向

3.1、消息保序 vs 負載均衡

預設選擇隊列的政策為輪詢方式，來保證消息可以均勻的配置設定到每個隊列；

既然說到隊列，就不得不提到消息的有序性問題

3.1.1、普通消息

消息是無序的，可發送至任意隊列，producer 也不關心消息會存儲在哪個隊列。在這種模式下，如果發送失敗，producer 會按照輪詢的方式，重新選取下一個隊列進行重試

producer.send(message);

3.1.2、普通有序消息

使用者可根據消息内容來選擇一個隊列發送 ，在這種情況下，消息也一般是保序的，例如我們可以通過業務字段（例如使用者id）的 msgKey 取模來選擇隊列，這樣同樣 msgKey 的消息必定會落在同一個隊列中。

與發送普通消息不同，如果發送失敗，将不會進行重試，也比較好了解，普通消息發送失敗後，也不會針對目前隊列進行重試，而是選擇下一個隊列

producer.send(zeroMsg, (mqs, msg, arg) -> {
    int index = msg.getKeys().hashCode() % mqs.size();
    return mqs.get(index);
}, 1000);

但也存在異常情況，例如目前 topic 的路由資訊發生了變化，取模後消息可能命中了另外一個隊列，自然也無法做到嚴格保序

3.1.3、嚴格有序消息

即 producer 自己嚴格發送給指定的隊列，如果發送異常則快速失敗，可見這種方式可以嚴格保證發送的消息在同一個隊列中，即便 topic 路由資訊發生變化，也可以嚴格保序

producer.send(message, messageQueue);

3.2、消息發送的3種方式

RocketMQ的rpc元件采用的是netty，而netty的網絡請求設計是完全異步的，是以一個請求避免一定可以拆成以下3個步驟

  • a、用戶端發送請求到伺服器（由于完全異步，是以請求資料可能隻放在了socket緩沖區，并沒有出網卡）
  • b、伺服器端處理請求（此過程不涉及網絡開銷，不過通常也是比較耗時的）
  • c、伺服器向用戶端傳回應答（請求的response）

3.2.1、同步發送消息 

SendResult result = producer.send(zeroMsg);

此過程比較好了解，即完成a、b、c所有步驟後才會傳回，耗時也是 a + b + c 的總和

3.2.2、異步發送消息

通常在業務中發送消息的代碼如下：

SendCallback sendCallback = new SendCallback() {
    @Override
    public void onSuccess(SendResult sendResult) {
        // doSomeThing;
    }
    @Override
    public void onException(Throwable e) {
        // doSomeThing;
    }
};
producer.send(zeroMsg, sendCallback);

而RocketMQ處理異步消息的邏輯是，直接啟動一個線程，而最終的結果異步回調

SendCallback

 
ExecutorService executor = this.getAsyncSenderExecutor();
try {
    executor.submit(new Runnable() {
        @Override
        public void run() {
            try {
                sendDefaultImpl(msg, CommunicationMode.ASYNC, sendCallback, timeout - costTime);
            } catch (Exception e) {
                sendCallback.onException(e);
            }
        }

    });
} catch (RejectedExecutionException e) {
    throw new MQClientException("executor rejected ", e);
}

3.2.2、單向發送消息 

producer.sendOneway(zeroMsg);

此模式與

sync

模式類似，都要經過producer端在資料發送前的資料組裝工作，不過在将資料交給netty，netty調用作業系統函數将資料放入socket緩沖區後，所有的過程便已結束。什麼場景會用到此模式呢？比如對可靠性要求并不高，但要求耗時非常短的場景，比如日志收集等

三個請求哪個更快呢？如果單論一個請求的話，肯定是

async

 異步的方式最快，因為它直接把工作交給另外一個線程去完成，主線程直接傳回了；但不論是

async

 還是

sync

 ，它們都是需要将 a、b、c 3個步驟都走完的，是以總開銷并不會減少。但

oneWay

 因為隻需将資料放入socket緩沖區後，client 端就直接傳回了，少了監聽并解析 server 端 response 的過程，是以可以得到最好的性能

四、總結

本章闡述了producer端相對重要的一些功能點，感覺比較核心的還是隊列相關的概念；但RocketMQ發展疊代了這麼多年，也涵蓋了很多及細小的特性，本文不能窮盡，比如“消息的壓縮”、“規避發送延遲較長的broker”、“逾時異常”等等，這些功能點獨立且零碎，讀源碼時可以帶着問題跟進，這樣針對性強，效率也會高很多

上一篇: HUAWEI網絡裝置恢複Console口密碼
下一篇: 模型壓縮明珠：二值化網絡