RocketMQ消息發送流程源碼剖析
正文
就是說,我們打瞭個比方,把RocketMQ比作碼頭上的一個小房子,來送孩子登船的傢長比作生產者,拉走孩子們的船夫比作消費者,所以,RocketMQ的故事就這麼展開瞭。
這節我們研究研究,消息的發送流程。也就是說,消息孩子從進門到坐到message queue座位上都經歷瞭啥。
父母把消息孩子送到碼頭之後,門口的門童defaultMQProducerImpl.send()
接過孩子,進入到MQ房子內部,然後引導孩子進入Broker
候船大廳內的message queue
座位上就坐。這就是消息發送的流程瞭。
而且孩子在剛被門童接到之後,就被規定瞭能在候船大廳待多久,默認是3秒。也就是說,要是再小房子內等瞭三秒沒走,就離開吧,你怕是沒想明白自己來幹啥的。這就是消息的超時時間。
讀源碼
1 調用defaultMQProducerImpl.send()
public SendResult send( Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { return send(msg, this.defaultMQProducer.getSendMsgTimeout()); }
2 設置過期時間
public SendResult send(Message msg, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { return this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout); }
3 執行defaultMQProducerImpl.sendDefaultImpl()方法
private SendResult sendDefaultImpl( Message msg, final CommunicationMode communicationMode, final SendCallback sendCallback, final long timeout ) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {}
這裡看看這幾個參數,
- communicationMode 是通信模式,同步異步還是單向
- sendCallback 是針對異步模式的,異步模式需要設置發送完成後的回調。
sendDefaultImpl是發送消息的核心方法。
這裡消息孩子進到第一個卡口,先要檢查送孩子來的傢長是否還能聯系上,若是能聯系到,就繼續。要是聯系不到,這孩子豈不是被拋棄瞭,不敢接不敢接,送到孤兒院吧。
然後需要檢查消息孩子瞭,首先是檢查孩子還在不在,別扔個衣服跑瞭。
然後看看孩子指定的這個topic,不能說我想去內個topic哈,必須是實實在在的名字。而且上頭也規定瞭,這個topic的名字也不能太長,也不能包含特殊字符。已有的一些領導定過的也不能用哈。
接下來就是檢查孩子的body瞭,之前說body就是孩子的技能,首先,技能為空,不行不行,啥都不會是不行的。再者太長也不行,你唱首歌兩年,這沒法玩。
檢查message不為null
檢查topic
- topic不能為空
- topic不能太長
- 不能包含特殊字符
檢查話題的名字是否被系統已占用
檢查body
- 檢查是否為空
- 檢查長度是否過長,最大為4MB 這樣
下邊我們看看sendDefaultImpl這個方法。給他拆成一段一段的看。
1 兩個校驗
//校驗生產者服務是ok的,可以聯系到的 this.makeSureStateOK(); //校驗消息的參數 Validators.checkMessage(msg, this.defaultMQProducer);
- 第一個檢查,檢查生產者服務是否是正常工作的,若是不正常工作,就拋出異常。
private void makeSureStateOK() throws MQClientException { if (this.serviceState != ServiceState.RUNNING) { throw new MQClientException("The producer service state not OK, " + this.serviceState + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK), null); } }
- 第二個檢查,檢查消息本身是否為空,檢查topic,檢查消息的body
public static void checkMessage(Message msg, DefaultMQProducer defaultMQProducer) throws MQClientException { if (null == msg) { throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message is null"); } // 這裡校驗Topic的時候,校驗瞭不能為空,長度和特殊字符 Validators.checkTopic(msg.getTopic()); //這裡則校驗瞭一些不允許使用的topic名字 Validators.isNotAllowedSendTopic(msg.getTopic()); // body不為空 if (null == msg.getBody()) { throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body is null"); } // body長度不為0 if (0 == msg.getBody().length) { throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body length is zero"); } // body 長度不能過長 if (msg.getBody().length > defaultMQProducer.getMaxMessageSize()) { throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body size over max value, MAX: " + defaultMQProducer.getMaxMessageSize()); } }
2 獲取topic路由信息
嗯,這裡孩子終於通過瞭檢查,服務人員開始帶著他去找自己指定的topic區域,指定是自己指定,劃分還是工作人員劃分的。咱總得知道這個topic區域在哪吧。
先去緩存筆記裡找,有沒有這個區域的信息,若是沒有這個topic,就新建一個,然後更新到緩存筆記裡邊。若有topic但是不知道在哪,就找name server
大腦去申請這個topic在哪的信息。
執行tryToFindTopicPublishInfo方法去獲取Topic的路由信息,若是不存在就新建,若是有topic但是緩存中沒有路由信息,則通過name server獲取路由信息。
TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) { //獲取topic信息 TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic); //不存在 if (null == topicPublishInfo || !topicPublishInfo.ok()) { //新建 this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo()); //修改topic的路由信息並更新到本地 this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic); topicPublishInfo = this.topicPublishInfoTable.get(topic); } //包含路由信息就直接返回 if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) { return topicPublishInfo; } else { //不包含路由信息則向name server申請,修改topic的路由信息並更新到本地 this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer); topicPublishInfo = this.topicPublishInfoTable.get(topic); return topicPublishInfo; } }
3 計算重試次數
這就是計算消息孩子可以嘗試去找地方坐幾次,沒坐上,欸,我又來瞭,沒坐上,欸,我又來瞭。
這行代碼就是計算重試次數的,根據communicationMode
傳入的值,同步異步還是單向的來決定重試次數是幾次。 很明顯,若是同步的,就會嘗試三次。若是異步的或者單向的就隻發送一次。
int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
4 執行隊列選擇方法
我們之前說瞭,Broker類似於候船大廳,為瞭均分壓力,每次都要進與上次不同的候船大廳。
執行selectOneMessageQueue
方法通過Queue將消息發送到與上次不同的一個Broker。也可以通過 sendLatencyFaultEnable判斷是否啟用延遲容錯開關
MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
5 發送消息
這就是走過巷道坐到屬於自己的座位上瞭
然後就通過sendKernelImpl
發送消息瞭,這是發送消息的核心方法。會準備通信層的入參,並將請求發送給通信層,內部實現是基於Netty
的。
sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
以上就是RocketMQ消息發送流程源碼剖析的詳細內容,更多關於RocketMQ消息發送流程的資料請關註WalkonNet其它相關文章!
推薦閱讀:
- java開發RocketMQ生產者高可用示例詳解
- 分佈式消息隊列RocketMQ概念詳解
- RocketMQ普通消息實戰演練詳解
- docker安裝RocketMQ的實現步驟
- RocketMQ消息隊列實現隨機消息發送當做七夕禮物