RocketMQ(二)：RPC通訊
阿新 • • 發佈：2018-04-06
clean router 意思 回車 實現 .cn 完整 換行 quest
匠心零度 轉載請註明原創出處，謝謝！
RocketMQ網絡部署圖
- NameServer：在系統中是做命名服務，更新和發現 broker服務。
- Broker-Master：broker 消息主機服務器。
- Broker-Slave： broker 消息從機服務器。
- Producer： 消息生產者。
- Consumer： 消息消費者。
rocketmq的幾個核心的模塊，而對於每個模塊都是單獨的jvm進程，我們看到上面的架構圖的時候，那些箭頭就是rocketmq的rpc調用，下面我們來看看rocketmq的rpc是如果進行封裝實現的。
說明： rocketmq系列都將會以rocketmq-4.1.0-incubating進行介紹。
RocketMQ通信組件
先排除Master、Slave直接通過原生的nio進行調用，其他通訊都是基於netty-all-4.0.36.Final以及RocketMQ自定義協議進行通訊的。
網絡協議定義如下
我們來看看header data裏面的數據定義：
code對於Request來說就是RequestCode類裏面的常量信息：
說明：公眾號【匠心零度】回復：rocketmq，可獲得基於rocketmq4.1.0加詳細中文代碼註釋 。
public class RequestCode {
// Broker 發送消息
public static final int SEND_MESSAGE = 10 code對於Response來說就是ResponseCode類裏面的常量信息：
public class ResponseCode extends RemotingSysResponseCode {
public static final int FLUSH_DISK_TIMEOUT = 10;
public static final int SLAVE_NOT_AVAILABLE = 11;
public static final int FLUSH_SLAVE_TIMEOUT = 12;
public static final int MESSAGE_ILLEGAL = 13;
public static final int SERVICE_NOT_AVAILABLE = 14;
public static final int VERSION_NOT_SUPPORTED = 15;
public static final int NO_PERMISSION = 16;
public static final int TOPIC_NOT_EXIST = 17;
public static final int TOPIC_EXIST_ALREADY = 18;
public static final int PULL_NOT_FOUND = 19;
public static final int PULL_RETRY_IMMEDIATELY = 20;
public static final int PULL_OFFSET_MOVED = 21;
public static final int QUERY_NOT_FOUND = 22;
public static final int SUBSCRIPTION_PARSE_FAILED = 23;
public static final int SUBSCRIPTION_NOT_EXIST = 24;
public static final int SUBSCRIPTION_NOT_LATEST = 25;
public static final int SUBSCRIPTION_GROUP_NOT_EXIST = 26;
public static final int FILTER_DATA_NOT_EXIST = 27;
public static final int FILTER_DATA_NOT_LATEST = 28;
public static final int TRANSACTION_SHOULD_COMMIT = 200;
public static final int TRANSACTION_SHOULD_ROLLBACK = 201;
public static final int TRANSACTION_STATE_UNKNOW = 202;
public static final int TRANSACTION_STATE_GROUP_WRONG = 203;
public static final int NO_BUYER_ID = 204;
public static final int NOT_IN_CURRENT_UNIT = 205;
public static final int CONSUMER_NOT_ONLINE = 206;
public static final int CONSUME_MSG_TIMEOUT = 207;
public static final int NO_MESSAGE = 208;
}flag字段進行說明，其他後續分析到具體的具體分析。
flag = 0表示是request，flag = 1表示是response。
private static final int RPC_TYPE = 0; // 0, REQUEST_COMMAND
public RemotingCommandType getType() {
if (this.isResponseType()) {//flag=1為true
return RemotingCommandType.RESPONSE_COMMAND;
}
return RemotingCommandType.REQUEST_COMMAND;
}
public boolean isResponseType() {
int bits = 1 << RPC_TYPE;
return (this.flag & bits) == bits;
}flag為2、3（二進制表示10、11）為oneway請求。
private static final int RPC_ONEWAY = 1; // Oneway bit
public void markOnewayRPC() {
int bits = 1 << RPC_ONEWAY;
this.flag |= bits;
}
public boolean isOnewayRPC() {
int bits = 1 << RPC_ONEWAY;
return (this.flag & bits) == bits;
}code=310很快我們就明白什麽意思了：
對於下面類似a、b、c等可以簡單查看下類SendMessageRequestHeaderV2（後續繼續講解）基本就是類似js壓縮效果，可以借鑒學習下。
public static SendMessageRequestHeaderV2 createSendMessageRequestHeaderV2(final SendMessageRequestHeader v1) {
SendMessageRequestHeaderV2 v2 = new SendMessageRequestHeaderV2();
/**
* 進行轉換，這樣網絡傳輸數據就比較小了，學習下
*/
v2.a = v1.getProducerGroup();
v2.b = v1.getTopic();
v2.c = v1.getDefaultTopic();
v2.d = v1.getDefaultTopicQueueNums();
v2.e = v1.getQueueId();
v2.f = v1.getSysFlag();
v2.g = v1.getBornTimestamp();
v2.h = v1.getFlag();
v2.i = v1.getProperties();
v2.j = v1.getReconsumeTimes();
v2.k = v1.isUnitMode();
v2.l = v1.getMaxReconsumeTimes();
v2.m = v1.isBatch();
return v2;
}備註： RemotingCommand類包含了傳輸過程中所有數據的封裝，還包括了編解碼等操作（非常棒！！！解讀為什麽這樣，從面向對象角度，誰擁有數據誰就對外提供操作這些數據的方法，這句話應該是大學的時候學習面向對象的時候看張孝祥老師說的，一直記憶猶新，的確應該這麽設計，rocketmq就這麽做的，再次學習）。
RocketMQ網絡協議實現
UML類圖
上面的圖已經做到非常清晰了，RemotingClient接口定義了client應該具備那些功能，RemotingSever類似，主要有：registerProcessor、invokeSync（同步調用）、invokeAsync（異步調用）、invokeOneway（單向調用）等等，而RemotingClient與RemotingSever在三種調用的區別就是參數有所區別。
NettyRemotingAbstract是Server與Client公用處理的抽象。
BrokerOuterAPI、MQClientImpl:都封裝了NettyRemotingClient（後續介紹）。
不管是client還是server通過RemotingService我們明白，啟動都是在start裏面，我們看看裏面核心netty代碼，以server裏面代碼為例：
備註：此處netty相關內容不進行深入展開，只會把涉及的的簡單說明，後續另開系列進行說明。
涉及主要ChannelHandler簡單說明
在進行tcp傳輸的時候經常會面臨黏包/拆包問題，netty自帶了很多通用的TCP黏包/拆包解決方案，下面我們看看rocketmq如何借助netty來實現編解碼：NettyEncoder編碼、NettyDecoder解碼，rocketmq相關的網絡協議上面內容已經說明過了。
NettyEncoder編碼
NettyDecoder解碼
netty中針對這四種場景均有對應的解碼器作為解決方案，比如：
- 通過FixedLengthFrameDecoder 定長解碼器來解決定長消息的黏包拆包問題。
- 通過LineBasedFrameDecoder和StringDecoder來解決以回車換行符作為消息結束符的TCP黏包拆包的問題。
- 通過DelimiterBasedFrameDecoder 特殊分隔符解碼器來解決以特殊符號作為消息結束符的TCP黏包拆包問題。
- 通過LengthFieldBasedFrameDecoder 自定義長度解碼器解決TCP黏包拆包問題。
rocketmq中使用的就是基於LengthFieldBasedFrameDecoder自定義長度解碼器的。
IdleStateHandler：Netty自帶的心跳檢測。
NettyConnetManageHandle：主要就是鏈接管理，新連接、連接斷開、異常、Idle等事件，每個事件過來存入NettyEventExecuter的隊列裏面。
NettyEventExecutor的run方法會不斷的從隊列裏面取事件進行相應的處理：
NettyServerHandler：具體業務處理（後續會說到）。
核心NettyRemotingAbstract介紹
invokeSync（同步調用）進行說明：
opaque就相當與標識的這個請求，雖然rpc調用請求發送結束了，但是響應回來的時候還是會帶有該信息就可以判斷出是原來那個請求，比如響應回來之後執行原來給定的回調等。
通過countDownLatch來控制等待網絡通信時間 ：
invokeAsync（異步調用）進行說明：
與invokeSync（同步調用）基本類似，boolean acquired = this.semaphoreAsync.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);//控制異步請求的個數以及超時和使用使用布爾原子變量，信號量保證只釋放一次，對於異步invokeCallback不為空，需要進行調用。invokeOneway（單向調用）比較簡單略過。
下面看看消息接收處理：
備註：這裏判斷是request還是response都是通過header裏面的flag標記來判斷的，上面已經說明。
processResponseCommand在介紹上面三種發送的時候說過了，下面重點看看processRequestCommand：
備註：這裏需要做流控，要求線程池對應的隊列必須是有大小限制的，是通過線程池進行限流的。
友情推薦
備註：http://www.iocoder.cn裏面有很多源碼系列分析，非常不錯，推薦給大家。
參考:
RocketMQ原理介紹V3.1.1
netty源碼分析之LengthFieldBasedFrameDecoder
如果讀完覺得有收獲的話，歡迎點贊、關註、加公眾號【匠心零度】，查閱更多精彩歷史！！！
加入知識星球，一起探討！
RocketMQ(二)：RPC通訊