如果你曾經(jīng)使用過RocketMQ，那么一定對(duì)以下發(fā)送消息的代碼不陌生
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);
Producer啟動(dòng)
其實(shí)僅僅一行代碼，在produer端的后臺(tái)啟動(dòng)了多個(gè)線程來協(xié)同工作，接下來我們逐一闡述
我們都知道，RocketMQ是一個(gè)集群部署、跨網(wǎng)絡(luò)的產(chǎn)品，除了producer、consumer需要網(wǎng)絡(luò)傳輸外，數(shù)據(jù)還需要在集群中流轉(zhuǎn)。所以一個(gè)高效、可靠的網(wǎng)絡(luò)組件是必不可少的。而RocketMQ選擇了netty
使用netty首先需要考慮的便是業(yè)務(wù)上的數(shù)據(jù)粘包問題，netty提供了一些較為常用的解決方案，如：固定長度（比如每次發(fā)送的消息長度均為1024byte）、固定分隔符等。而RocketMQ使用的則是最為通用的head、body分離方式，即head存儲(chǔ)消息的長度，body存儲(chǔ)真正的消息數(shù)據(jù)，具體實(shí)現(xiàn)可參見類o.a.r.r.n.NettyRemotingClient
而消息收發(fā)這塊，RocketMQ將所有的消息都收斂到同一個(gè)協(xié)議類o.a.r.r.p.RemotingCommand中，即消息發(fā)送、接收都會(huì)將其封裝在該類中，這樣做的好處是不言而喻的，即統(tǒng)一規(guī)范，減輕網(wǎng)絡(luò)協(xié)議適配不同的消息類型帶來的負(fù)擔(dān)
其中較為重要的2個(gè) ChannelHanlder 如下
org.apache.rocketmq.remoting.netty.NettyEncoder
消息編碼，向 broker 或 nameServer 發(fā)送消息時(shí)使用，將RemotingCommand轉(zhuǎn)換為byte[]形式
org.apache.rocketmq.remoting.netty.NettyDecoder
消息解碼，將byte[]轉(zhuǎn)換為RemotingCommand對(duì)象，接收 broker 返回的消息時(shí)，進(jìn)行解碼操作
消息格式
消息格式是什么概念？在《消息存儲(chǔ)》章節(jié)不是已經(jīng)闡述過消息格式了嗎？其實(shí)這是兩個(gè)概念，《消息存儲(chǔ)》章節(jié)是消息真正落盤時(shí)候的存儲(chǔ)格式，本小節(jié)的消息格式是指消息以什么樣的形態(tài)交給netty從而在網(wǎng)絡(luò)上進(jìn)行傳輸
消息格式
消息格式是什么概念？在《消息存儲(chǔ)》章節(jié)不是已經(jīng)闡述過消息格式了嗎？其實(shí)這是兩個(gè)概念，《消息存儲(chǔ)》章節(jié)是消息真正落盤時(shí)候的存儲(chǔ)格式，本小節(jié)的消息格式是指消息以什么樣的形態(tài)交給netty從而在網(wǎng)絡(luò)上進(jìn)行傳輸
消息格式由MsgHeader及MsgBody組成，而消息的長度、標(biāo)記、版本等重要參數(shù)都放在 header 中，body 中僅僅存儲(chǔ)數(shù)據(jù)，沒有額外字段；我們主要看一下 header 的數(shù)據(jù)格式
而站在 netty 視角來看，不論是 msgHeader 還是 msgBody，都屬于 netty 網(wǎng)絡(luò)消息的body部分，所以我們可以簡單畫一張 netty 視角的消息格式
Msg Header的自動(dòng)適配
上文得知，RocketMQ將所有的消息類型、收發(fā)都收斂到類RemotingCommand中，但RocketMQ消息類型眾多，除了常見的消息發(fā)送、接收外，還有通過msgID查詢消息、msgKey查詢消息、獲取broker配置、清理不再使用的topic等等，用一個(gè)類適配如此多的類型，具體是如何實(shí)現(xiàn)的呢？當(dāng)新增、修改一種類型又該怎么應(yīng)對(duì)呢？
翻看源碼便發(fā)現(xiàn)，RemotingCommand的消息頭定義為一個(gè)接口org.apache.rocketmq.remoting.CommandCustomHeader，不同類型的請(qǐng)求都實(shí)現(xiàn)這個(gè)接口，并在自己的子類中定義成員變量；那RemotingCommand的消息頭又是如何自動(dòng)解析呢？
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()通過犧牲少量性能的方式，換取了程序極大的靈活性與擴(kuò)展性，當(dāng)新增請(qǐng)求類型時(shí)，僅需要編寫新請(qǐng)求的encode、decode，不用修改其他類型請(qǐng)求的代碼
Topic路由信息
Topic創(chuàng)建
發(fā)送消息的前置是需要?jiǎng)?chuàng)建一個(gè)topic，創(chuàng)建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
簡單介紹下每個(gè)參數(shù)的作用
-b broker 地址，表示 topic 所在 Broker，只支持單臺(tái)Broker，地址為ip:port
-c cluster 地址，表示 topic 所在 cluster，會(huì)向 cluster 中所有的 broker 發(fā)送請(qǐng)求
-t topic 名稱
-r 可讀隊(duì)列數(shù)（默認(rèn)為 8，后文還會(huì)展開）
-w 可寫隊(duì)列數(shù)（默認(rèn)為 8，后文還會(huì)展開）
-p 指定新topic的讀寫權(quán)限 （W=2|R=4|WR=6）2表示當(dāng)前topic僅可寫入數(shù)據(jù)，4表示僅可讀，6表示可讀可寫
-o set topic's order(true|false)
-u is unit topic (true|false)
-s has unit sub (true|false)
如果執(zhí)行命令updateTopic -b 127.0.0.1:8899 -t testTopic -r 8 -w 8 意味著會(huì)在127.0.0.1:8899對(duì)應(yīng)的broker下創(chuàng)建一個(gè)topic，這個(gè)topic的讀寫隊(duì)列都是 8
那如果是這樣的場(chǎng)景呢：集群A有3個(gè)master節(jié)點(diǎn)，當(dāng)執(zhí)行命令updateTopic -c clusterName -t testTopic -r 8 -w 8 后，站在集群A角度來看，當(dāng)前topic總共創(chuàng)建了多少個(gè)寫隊(duì)列？其實(shí) RocketMQ 接到這條命令后，會(huì)向3個(gè) broker 分別發(fā)送創(chuàng)建 topic 的命令，這樣每個(gè)broker上都會(huì)有8個(gè)讀隊(duì)列，8個(gè)寫隊(duì)列，所以站在集群的視角，這個(gè) topic 總共會(huì)有 24 個(gè)讀隊(duì)列，24 個(gè)寫隊(duì)列
創(chuàng)建流程
一、創(chuàng)建Topic的客戶端（DefaultMQAdminExt）
第一步：該客戶端的啟動(dòng)流程與Producer、Consumer類似，需要start()，它們共用MQClientInstance#start()方法，啟動(dòng)后還有多個(gè)后臺(tái)輪訓(xùn)線程
第二步：通過與NameServer交互，將指定ClusterName下所有的Broker信息拉下來
第三步：依次向這些Broker發(fā)送創(chuàng)建Topic的請(qǐng)求
二、Broker
第一步：Broker收到創(chuàng)建Topic的請(qǐng)求后，做一些新Topic的初始化動(dòng)作，而后將該Topic的元數(shù)據(jù)存儲(chǔ)在一個(gè)name為topics.json的本地文件中，因?yàn)樵贜ameServer中并沒有對(duì)數(shù)據(jù)進(jìn)行持久化，所以此文件即為Topic路由數(shù)據(jù)的唯一持久化文件，當(dāng)然這樣的Broker一般是有多套的（其實(shí)此處是將所有json數(shù)據(jù)全部實(shí)例化后，替換本地文件，真實(shí)生產(chǎn)中，如果頻繁創(chuàng)建、銷毀topic，會(huì)帶來大量的文件IO，以及內(nèi)存負(fù)擔(dān)，相信在未來近期的某個(gè)版本一定會(huì)進(jìn)行修復(fù)）
第二步：向所有NameServer列表挨個(gè)發(fā)送Topic注冊(cè)請(qǐng)求
三、NameServer
NameServer收到Broker注冊(cè)Topic的消息后，便將其路由信息存儲(chǔ)在內(nèi)存中，當(dāng)有Client請(qǐng)求Topic路由數(shù)據(jù)時(shí)，便將結(jié)果同步過去
我們以3個(gè)Broker、2個(gè)NameServer的集群舉例：
Client → Broker ：3 次網(wǎng)絡(luò)IO，Client需要挨個(gè)向多個(gè)Broker發(fā)送注冊(cè)請(qǐng)求
Broker → NameServer：6 次網(wǎng)絡(luò)IO，Broker需要向所有NameServer發(fā)送注冊(cè)請(qǐng)求
由此可見，NameServer確實(shí)是輕狀態(tài)的節(jié)點(diǎn)，路由的原始數(shù)據(jù)其實(shí)都存儲(chǔ)在Broker上，通過Broker向NameServer注冊(cè)，再有Client從NameServer處獲取元數(shù)據(jù)的方式來進(jìn)行廣播、同步。此方案是rmq獨(dú)創(chuàng)，與kafka的重ZooKeeper形成對(duì)比，不過從實(shí)踐角度看，該架構(gòu)還是比較穩(wěn)定的
writeQueueNum VS readQueueNum
首選需要明確的是，讀、寫隊(duì)列，這兩個(gè)概念是 RocketMQ 獨(dú)有的，而 kafka 中只有一個(gè)partition的概念，不區(qū)分讀寫。一般情況下，這兩個(gè)值建議設(shè)置為相等；我們分別看一下 client 端對(duì)它們的處理 （均在類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個(gè)隊(duì)列設(shè)置不相等，例如我們?cè)O(shè)置6個(gè)寫隊(duì)列，4個(gè)讀隊(duì)列的話：
這樣，4、5號(hào)隊(duì)列中的數(shù)據(jù)一定不會(huì)被消費(fèi)。
writeQueueNum > readQueueNum
大于 readQueueNum 部分的隊(duì)列永遠(yuǎn)不會(huì)被消費(fèi)
writeQueueNum < readQueueNum
所有隊(duì)列中的數(shù)據(jù)都會(huì)被消費(fèi)，但部分讀隊(duì)列數(shù)據(jù)一直是空的
這樣設(shè)計(jì)有什么好處呢？其實(shí)是更精細(xì)的控制了讀寫操作，例如當(dāng)我們要遷移 broker 時(shí)，可以首先將寫入隊(duì)列設(shè)置為0，將客戶端引流至其他 broker 節(jié)點(diǎn)，等讀隊(duì)列數(shù)據(jù)也處理完畢后，再關(guān)閉 read 操作
路由數(shù)據(jù)格式
topic的路由數(shù)據(jù)如何由Admin發(fā)起創(chuàng)建，再被各個(gè)broker響應(yīng)，繼而被nameServer統(tǒng)一組織創(chuàng)建的流程我們暫且不討論，為防止發(fā)散，我們直接從producer從nameServer獲取路由數(shù)據(jù)開始。從nameServer獲取到的路由數(shù)據(jù)格式如下
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;
}
而存放路由數(shù)據(jù)的結(jié)構(gòu)是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的概念
上圖描述了一個(gè)cluster下有3個(gè)broker，每個(gè)broker又有1個(gè)master，2個(gè)slave組成；這也就是為什么類BrokerData中有HashMap<Long, String> brokerAddrs變量的原因，因?yàn)榭赡芡粋€(gè)brokerName下由多個(gè)節(jié)點(diǎn)組成。注：master節(jié)點(diǎn)的編號(hào)始終為0
Topic路由信息何時(shí)發(fā)生變化
這些路由信息什么時(shí)候發(fā)生變化呢？我們舉例說明
舉例1：某集群有3臺(tái) master，分別向其中的2臺(tái)發(fā)送了創(chuàng)建topic的命令，此時(shí)所有的clent端都知道這個(gè)topic的數(shù)據(jù)在這兩個(gè)broker上；這個(gè)時(shí)候通過admin向第3臺(tái)broker發(fā)送創(chuàng)建topic命令，nameServer的路由信息便發(fā)生了變更，等client端30秒輪訓(xùn)后，便可以更新到最新的topic路由信息
舉例2：某集群有3臺(tái) master，topic分別在3臺(tái)broker上都創(chuàng)建了，此時(shí)某臺(tái)broker宕機(jī)，nameServer將其摘除，等待30秒輪詢后，client拿到最新路由信息
思考：client 端路由信息的變化是依托于30秒的輪詢，如果路由信息已經(jīng)發(fā)生變化，且輪詢未發(fā)生，client端拿著舊的topic路由信息訪問集群，一定會(huì)有短暫報(bào)錯(cuò)，此處也是待優(yōu)化的點(diǎn)
定時(shí)更新Topic路由信息
RocketMQ會(huì)每隔30秒更新topic的路由信息
與Broker心跳
主要分為兩部分：
1、清空無效broker
2、向有效的broker發(fā)送心跳
2.4.1、清空無效的broker
由上節(jié)得知，RocketMQ會(huì)獲取所有已經(jīng)注冊(cè)的topic所在的broker信息，并將這些信息存儲(chǔ)在變量brokerAddrTable中，brokerAddrTable的存儲(chǔ)結(jié)構(gòu)如下
key: brokerName，例如一個(gè)master帶2個(gè)slave都屬于同一個(gè)brokerName
val: HashMap<Long, String>，key為brokerId（其中master的brokerId固定為0），val為ip地址
如何判斷某個(gè)broker有效無效呢？判斷依據(jù)便是MQClientInstance#topicRouteTable，這個(gè)變量是上節(jié)中從nameServer中同步過來的，如果brokerAddrTable中有broker A、B、C，而topicRouteTable只有A、B的話，那么就需要從brokerAddrTable中刪除C。
需要注意的是，在整個(gè)check及替換過程中都添加了獨(dú)占鎖lockNamesrv，而上節(jié)中維護(hù)更新topic路由信息也是指定的該鎖
發(fā)送心跳數(shù)據(jù)
此處目的僅為與broker保持網(wǎng)絡(luò)心跳，如果連接失敗或發(fā)生異常，僅會(huì)打印日志，并不會(huì)有額外操作
多Producer
這里簡單提一下，其實(shí)在單個(gè)進(jìn)程中，是可以啟動(dòng)多個(gè)Producer的，且相互隔離；實(shí)現(xiàn)起來感覺也比較容易，感覺直接new DefaultMQProducer()就行。不過這里有個(gè)性能上的問題，就是如果兩個(gè)Producer操作了同樣的Topic，此時(shí)去NameServer拉取路由數(shù)據(jù)的時(shí)候，將會(huì)線性的放大，因此RMQ引入了MQClientInstance概念，即在單個(gè)進(jìn)程中，MQClientInstance是單例的，諸如獲取Topic路由數(shù)據(jù)等，均是其統(tǒng)一發(fā)起，讀者在源碼中看到這個(gè)類時(shí)不要覺得陌生哈
消息發(fā)送
消息發(fā)送比較重要的是2點(diǎn)內(nèi)容
發(fā)送數(shù)據(jù)的負(fù)載均衡問題；RocketMQ默認(rèn)采用的是輪訓(xùn)的方式
消息發(fā)送的方式；分同步、異步、單向
消息保序 vs 負(fù)載均衡
默認(rèn)選擇隊(duì)列的策略為輪詢方式，來保證消息可以均勻的分配到每個(gè)隊(duì)列；
既然說到隊(duì)列，就不得不提到消息的有序性問題
普通消息
消息是無序的，可發(fā)送至任意隊(duì)列，producer 也不關(guān)心消息會(huì)存儲(chǔ)在哪個(gè)隊(duì)列。在這種模式下，如果發(fā)送失敗，producer 會(huì)按照輪詢的方式，重新選取下一個(gè)隊(duì)列進(jìn)行重試
producer.send(message);
普通有序消息
用戶可根據(jù)消息內(nèi)容來選擇一個(gè)隊(duì)列發(fā)送 ，在這種情況下，消息也一般是保序的，例如我們可以通過業(yè)務(wù)字段（例如用戶id）的 msgKey 取模來選擇隊(duì)列，這樣同樣 msgKey 的消息必定會(huì)落在同一個(gè)隊(duì)列中。
與發(fā)送普通消息不同，如果發(fā)送失敗，將不會(huì)進(jìn)行重試，也比較好理解，普通消息發(fā)送失敗后，也不會(huì)針對(duì)當(dāng)前隊(duì)列進(jìn)行重試，而是選擇下一個(gè)隊(duì)列
producer.send(zeroMsg, (mqs, msg, arg) -> {
int index = msg.getKeys().hashCode() % mqs.size();
return mqs.get(index);
}, 1000);
但也存在異常情況，例如當(dāng)前 topic 的路由信息發(fā)生了變化，取模后消息可能命中了另外一個(gè)隊(duì)列，自然也無法做到嚴(yán)格保序
嚴(yán)格有序消息
即 producer 自己嚴(yán)格發(fā)送給指定的隊(duì)列，如果發(fā)送異常則快速失敗，可見這種方式可以嚴(yán)格保證發(fā)送的消息在同一個(gè)隊(duì)列中，即便 topic 路由信息發(fā)生變化，也可以嚴(yán)格保序
producer.send(message, messageQueue);
消息發(fā)送的3種方式
RocketMQ的rpc組件采用的是netty，而netty的網(wǎng)絡(luò)請(qǐng)求設(shè)計(jì)是完全異步的，所以一個(gè)請(qǐng)求一定可以拆成以下3個(gè)步驟
a、客戶端發(fā)送請(qǐng)求到服務(wù)器（由于完全異步，所以請(qǐng)求數(shù)據(jù)可能只放在了socket緩沖區(qū)，并沒有出網(wǎng)卡）
b、服務(wù)器端處理請(qǐng)求（此過程不涉及網(wǎng)絡(luò)開銷，不過通常也是比較耗時(shí)的）
c、服務(wù)器向客戶端返回應(yīng)答（請(qǐng)求的response）
同步發(fā)送消息
SendResult result = producer.send(zeroMsg);
此過程比較好理解，即完成a、b、c所有步驟后才會(huì)返回，耗時(shí)也是 a + b + c 的總和
3.2.2、異步發(fā)送消息
通常在業(yè)務(wù)中發(fā)送消息的代碼如下：
SendCallback sendCallback = new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
// doSomeThing;
}
@Override
public void onException(Throwable e) {
// doSomeThing;
}
};
producer.send(zeroMsg, sendCallback);
而RocketMQ處理異步消息的邏輯是，直接啟動(dòng)一個(gè)線程，而最終的結(jié)果異步回調(diào)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);
}
單向發(fā)送消息
producer.sendOneway(zeroMsg);
此模式與sync模式類似，都要經(jīng)過producer端在數(shù)據(jù)發(fā)送前的數(shù)據(jù)組裝工作，不過在將數(shù)據(jù)交給netty，netty調(diào)用操作系統(tǒng)函數(shù)將數(shù)據(jù)放入socket緩沖區(qū)后，所有的過程便已結(jié)束。什么場(chǎng)景會(huì)用到此模式呢？比如對(duì)可靠性要求并不高，但要求耗時(shí)非常短的場(chǎng)景，比如日志收集等
三個(gè)請(qǐng)求哪個(gè)更快呢？如果單論一個(gè)請(qǐng)求的話，肯定是async異步的方式最快，因?yàn)樗苯影压ぷ鹘唤o另外一個(gè)線程去完成，主線程直接返回了；但不論是async還是sync，它們都是需要將 a、b、c 3個(gè)步驟都走完的，所以總開銷并不會(huì)減少。但oneWay因?yàn)橹恍鑼?shù)據(jù)放入socket緩沖區(qū)后，client 端就直接返回了，少了監(jiān)聽并解析 server 端 response 的過程，所以可以得到最好的性能
小結(jié)
本文闡述了producer端相對(duì)重要的一些功能點(diǎn)，感覺比較核心的還是隊(duì)列相關(guān)的概念；但RocketMQ發(fā)展迭代了這么多年，也涵蓋了很多及細(xì)小的特性，本文不能窮盡，比如“消息的壓縮”、“規(guī)避發(fā)送延遲較長的broker”、“超時(shí)異?！钡鹊?，這些功能點(diǎn)獨(dú)立且零碎，讀源碼時(shí)可以帶著問題跟進(jìn)，這樣針對(duì)性強(qiáng)，效率也會(huì)高很多
