RocketMQ訊息傳送流程原始碼剖析

2022-08-01 18:05:45

正文

就是說，我們打了個比方，把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訊息傳送流程的資料請關注it145.com其它相關文章！