轉至 ：匠心零度
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類裡面的常量信息：
public class RequestCode { // Broker 發送消息 public static final int SEND_MESSAGE = 10; // Broker 訂閱消息 public static final int PULL_MESSAGE = 11; // Broker 查詢消息 public static final int QUERY_MESSAGE = 12; // Broker 查詢Broker Offset public static final int QUERY_BROKER_OFFSET = 13; // Broker 查詢Consumer Offset public static final int QUERY_CONSUMER_OFFSET = 14; // Broker 更新Consumer Offset public static final int UPDATE_CONSUMER_OFFSET = 15; // Broker 更新或者增加一個Topic public static final int UPDATE_AND_CREATE_TOPIC = 17; // Broker 獲取所有Topic的配置（Slave和Namesrv都會向Master請求此配置） public static final int GET_ALL_TOPIC_CONFIG = 21; // Broker 獲取所有Topic配置（Slave和Namesrv都會向Master請求此配置 public static final int GET_TOPIC_CONFIG_LIST = 22; // Broker 獲取所有Topic名稱列表 public static final int GET_TOPIC_NAME_LIST = 23; // Broker 更新Broker上的配置 public static final int UPDATE_BROKER_CONFIG = 25; // Broker 獲取Broker上的配置 public static final int GET_BROKER_CONFIG = 26; // Broker 觸發Broker刪除文件 public static final int TRIGGER_DELETE_FILES = 27; // Broker 獲取Broker運行時信息 public static final int GET_BROKER_RUNTIME_INFO = 28; // Broker 根據時間查詢隊列的Offset public static final int SEARCH_OFFSET_BY_TIMESTAMP = 29; // Broker 查詢隊列最大Offset public static final int GET_MAX_OFFSET = 30; // Broker 查詢隊列最小Offset public static final int GET_MIN_OFFSET = 31; // Broker 查詢隊列最早消息對應時間 public static final int GET_EARLIEST_MSG_STORETIME = 32; // Broker 根據消息ID來查詢消息 public static final int VIEW_MESSAGE_BY_ID = 33; // Broker Client向Client發送心跳，並註冊自身 public static final int HEART_BEAT = 34; // Broker Client註銷 public static final int UNREGISTER_CLIENT = 35; // Broker Consumer將處理不了的消息發回服務器 public static final int CONSUMER_SEND_MSG_BACK = 36; // Broker Commit或者Rollback事務 public static final int END_TRANSACTION = 37; // Broker 獲取ConsumerId列表通過GroupName public static final int GET_CONSUMER_LIST_BY_GROUP = 38; // Broker 主動向Producer回查事務狀態 public static final int CHECK_TRANSACTION_STATE = 39; // Broker Broker通知Consumer列表變化 public static final int NOTIFY_CONSUMER_IDS_CHANGED = 40; // Broker Consumer向Master鎖定隊列 public static final int LOCK_BATCH_MQ = 41; // Broker Consumer向Master解鎖隊列 public static final int UNLOCK_BATCH_MQ = 42; // Broker 獲取所有Consumer Offset public static final int GET_ALL_CONSUMER_OFFSET = 43; // Broker 獲取所有定時進度 public static final int GET_ALL_DELAY_OFFSET = 45; public static final int CHECK_CLIENT_CONFIG = 46; // Namesrv 向Namesrv追加KV配置 public static final int PUT_KV_CONFIG = 100; // Namesrv 從Namesrv獲取KV配置 public static final int GET_KV_CONFIG = 101; // Namesrv 從Namesrv獲取KV配置 public static final int DELETE_KV_CONFIG = 102; // Namesrv 註冊一個Broker，數據都是持久化的，如果存在則覆蓋配置 public static final int REGISTER_BROKER = 103; // Namesrv 卸載一個Broker，數據都是持久化的 public static final int UNREGISTER_BROKER = 104; // Namesrv 根據Topic獲取Broker Name、隊列數(包含讀隊列與寫隊列) public static final int GET_ROUTEINTO_BY_TOPIC = 105; // Namesrv 獲取註冊到Name Server的所有Broker集群信息 public static final int GET_BROKER_CLUSTER_INFO = 106; public static final int UPDATE_AND_CREATE_SUBSCRIPTIONGROUP = 200; public static final int GET_ALL_SUBSCRIPTIONGROUP_CONFIG = 201; public static final int GET_TOPIC_STATS_INFO = 202; public static final int GET_CONSUMER_CONNECTION_LIST = 203; public static final int GET_PRODUCER_CONNECTION_LIST = 204; public static final int WIPE_WRITE_PERM_OF_BROKER = 205; // 從Name Server獲取完整Topic列表 public static final int GET_ALL_TOPIC_LIST_FROM_NAMESERVER = 206; // 從Broker刪除訂閱組 public static final int DELETE_SUBSCRIPTIONGROUP = 207; // 從Broker獲取消費狀態（進度） public static final int GET_CONSUME_STATS = 208; // Suspend Consumer消費過程 public static final int SUSPEND_CONSUMER = 209; // Resume Consumer消費過程 public static final int RESUME_CONSUMER = 210; // 重置Consumer Offset public static final int RESET_CONSUMER_OFFSET_IN_CONSUMER = 211; // 重置Consumer Offset public static final int RESET_CONSUMER_OFFSET_IN_BROKER = 212; // 調整Consumer線程池數量 public static final int ADJUST_CONSUMER_THREAD_POOL = 213; // 查詢消息被哪些消費組消費 public static final int WHO_CONSUME_THE_MESSAGE = 214; // 從Broker刪除Topic配置 public static final int DELETE_TOPIC_IN_BROKER = 215; // 從Namesrv刪除Topic配置 public static final int DELETE_TOPIC_IN_NAMESRV = 216; // 通過NameSpace獲取所有的KV List public static final int GET_KVLIST_BY_NAMESPACE = 219; // offset 重置 public static final int RESET_CONSUMER_CLIENT_OFFSET = 220; // 客戶端訂閱消息 public static final int GET_CONSUMER_STATUS_FROM_CLIENT = 221; // 通知 broker 調用 offset 重置處理 public static final int INVOKE_BROKER_TO_RESET_OFFSET = 222; // 通知 broker 調用客戶端訂閱消息處理 public static final int INVOKE_BROKER_TO_GET_CONSUMER_STATUS = 223; // Broker 查詢topic被誰消費 public static final int QUERY_TOPIC_CONSUME_BY_WHO = 300; // 獲取指定集群下的所有 topic public static final int GET_TOPICS_BY_CLUSTER = 224; // 向Broker註冊Filter Server public static final int REGISTER_FILTER_SERVER = 301; // 向Filter Server註冊Class public static final int REGISTER_MESSAGE_FILTER_CLASS = 302; // 根據 topic 和 group 獲取消息的時間跨度 public static final int QUERY_CONSUME_TIME_SPAN = 303; // 獲取所有系統內置 Topic 列表 public static final int GET_SYSTEM_TOPIC_LIST_FROM_NS = 304; public static final int GET_SYSTEM_TOPIC_LIST_FROM_BROKER = 305; // 清理失效隊列 public static final int CLEAN_EXPIRED_CONSUMEQUEUE = 306; // 通過Broker查詢Consumer內存數據 public static final int GET_CONSUMER_RUNNING_INFO = 307; // 查找被修正 offset (轉發組件） public static final int QUERY_CORRECTION_OFFSET = 308; // 通過Broker直接向某個Consumer發送一條消息，並立刻消費，返回結果給broker，再返回給調用方 public static final int CONSUME_MESSAGE_DIRECTLY = 309; // Broker 發送消息，優化網絡數據包 public static final int SEND_MESSAGE_V2 = 310; // 單元化相關 topic public static final int GET_UNIT_TOPIC_LIST = 311; // 獲取含有單元化訂閱組的 Topic 列表 public static final int GET_HAS_UNIT_SUB_TOPIC_LIST = 312; // 獲取含有單元化訂閱組的非單元化 Topic 列表 public static final int GET_HAS_UNIT_SUB_UNUNIT_TOPIC_LIST = 313; // 克隆某一個組的消費進度到新的組 public static final int CLONE_GROUP_OFFSET = 314; // 查看Broker上的各種統計信息 public static final int VIEW_BROKER_STATS_DATA = 315; public static final int CLEAN_UNUSED_TOPIC = 316; public static final int GET_BROKER_CONSUME_STATS = 317; /** * update the config of name server */ public static final int UPDATE_NAMESRV_CONFIG = 318; /** * get config from name server */ public static final int GET_NAMESRV_CONFIG = 319; public static final int SEND_BATCH_MESSAGE = 320; public static final int QUERY_CONSUME_QUEUE = 321;} 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_COMMANDpublic 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 bitpublic 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：
備註：這裡需要做流控，要求線程池對應的隊列必須是有大小限制的，是通過線程池進行限流的。
閱讀更多 java進階架構師 的文章
