RocketMQ—Producer（四）消息發(fā)送流程

前 言

在深入講解消息發(fā)送之前，我們可先簡單概括消息的發(fā)送的主要步驟可分為：消息驗證、路由查詢、選擇消息隊列、消息組裝、消息發(fā)送、消息結(jié)果處理、異常處理；(單向發(fā)送并不處理消息發(fā)送結(jié)果)；同步、異步、單向發(fā)送消息的入口API有一些區(qū)別，本文將以下面接口實現(xiàn)類為入口分析消息發(fā)送的流程：

DefaultMQProducerImpl#sendDefaultImpl

(由于消息發(fā)送細節(jié)非常多，本文將分析核心步驟，如漏掉還請各位查漏補缺，自行分析哈)

同步發(fā)送總結(jié)流程圖如下：

image.png

一、源碼分析

DefaultMQProducerImpl#sendDefaultImpl

/**
 * 發(fā)送信息
 * @param msg 消息內(nèi)容
 * @param communicationMode 發(fā)送模式
 * @param sendCallback 回掉
 * @param timeout 超時時間
 */
private SendResult sendDefaultImpl(
    Message msg,
    final CommunicationMode communicationMode,
    final SendCallback sendCallback,
    final long timeout
) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
    this.makeSureStateOK();         //驗證 serviceState == Running 運行中
    Validators.checkMessage(msg, this.defaultMQProducer);   //1> 驗證消息

    final long invokeID = random.nextLong();//隨機的-invokeId
    long beginTimestampFirst = System.currentTimeMillis();//開始時間
    long beginTimestampPrev = beginTimestampFirst;
    long endTimestamp = beginTimestampFirst;
    TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic()); // 2> 獲取路由信息
    if (topicPublishInfo != null && topicPublishInfo.ok()) {
        boolean callTimeout = false;
        MessageQueue mq = null;
        Exception exception = null;
        SendResult sendResult = null;
        int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;//重試次數(shù)，同步默認3，其他1次
        int times = 0;
        String[] brokersSent = new String[timesTotal];//發(fā)送的brokerName集合
        for (; times < timesTotal; times++) {
            String lastBrokerName = null == mq ? null : mq.getBrokerName();
            MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName); // 3>選擇消息隊列
            if (mqSelected != null) {
                mq = mqSelected;
                brokersSent[times] = mq.getBrokerName();
                try {
                    beginTimestampPrev = System.currentTimeMillis();//本次開始時間
                    long costTime = beginTimestampPrev - beginTimestampFirst;//計算發(fā)送消耗時間
                    if (timeout < costTime) {//如果消耗時間 大于 超時時間，直接break
                        callTimeout = true;
                        break;
                    }
                    //發(fā)送消息
                    sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime); // 4>消息發(fā)送
                    //發(fā)送完成時間
                    endTimestamp = System.currentTimeMillis();
                    //更新失敗條目信息
                    this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
                    switch (communicationMode) {
                        case ASYNC:
                            return null;
                        case ONEWAY:
                            return null;
                        case SYNC:
                            if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
                                if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
                                    continue;
                                }
                            }

                            return sendResult;
                        default:
                            break;
                    }
                } catch (RemotingException e) {
                    endTimestamp = System.currentTimeMillis();
                    this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);  //5>更新失敗條目
                    log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
                    log.warn(msg.toString());
                    exception = e;
                    continue;
                    ...省略...
            } else {    //沒有找到消息隊列，直接break
                break;
            }
        }
        if (sendResult != null) {
            return sendResult;
        }
        String info = String.format("Send [%d] times, still failed, cost [%d]ms, Topic: %s, BrokersSent: %s",
            times,
            System.currentTimeMillis() - beginTimestampFirst,
            msg.getTopic(),
            Arrays.toString(brokersSent));

        info += FAQUrl.suggestTodo(FAQUrl.SEND_MSG_FAILED);

        MQClientException mqClientException = new MQClientException(info, exception);
            ...省略...
        throw mqClientException;
    }
   ...省略...
}

1.1 驗證消息

Validators.checkMessage

//Validate message 驗證消息
public static void checkMessage(Message msg, DefaultMQProducer defaultMQProducer)
    throws MQClientException {
    if (null == msg) {
        throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message is null");
    }
    Validators.checkTopic(msg.getTopic()); // 驗證topic, 此處代碼大家可自行查看，灰常簡單
    if (null == msg.getBody()) {           // body 消息體不能為空
        throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body is null");
    }
    if (0 == msg.getBody().length) {
        throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body length is zero");
    }
    //消息最大長度 不能大于 4M
    if (msg.getBody().length > defaultMQProducer.getMaxMessageSize()) {
        throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL,
            "the message body size over max value, MAX: " + defaultMQProducer.getMaxMessageSize());
    }
}

備注:

主要驗證消息分為兩部分

  1. topic驗證:匹配正則表達式(^[%|a-zA-Z0-9_-]+$),長度小于255,不等于默認主題:TBW102
  2. body驗證:body內(nèi)容是否為空，消息內(nèi)容最大長度默認不能超過4M

1.2 獲取路由信息

tryToFindTopicPublishInfo

在'路由動態(tài)更新'我們以及分析過了，代碼大家可以再回顧下，簡單邏輯總結(jié)如下:

  1. 如果生產(chǎn)者中緩存了 topic 的路由信息，如果該路由信息中包含了消息隊列，則直接返回該路由信息;
  2. 如果沒有緩存或沒有包含消息隊列， 則向 NameServer查詢該 topic 的路由信息;
  3. 如果最終未找到路由信息，則拋出異常 : 無法找到主題相關(guān)路由信息異常.

1.3 選擇消息隊列

將在'系列5'著重分析此段代碼功能消息

1.4 消息發(fā)送

sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);

由于代碼篇幅太長,下面講解只摘取sendKernelImpl方法的核心代碼解析,但強烈建議仔細去擼一遍代碼消息。

1.4.1 查詢-brokerAddr

String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
if (null == brokerAddr) {
    tryToFindTopicPublishInfo(mq.getTopic());
    brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
}
if(brokerAddr != null) {
  ... 省略 ...
} else{
  拋異常
}

邏輯：

從brokerAddrTable獲取主MasterId,獲取不到則查詢路由，如果繼續(xù)獲取不到則跑異常消息

 //MQClientInstance#findBrokerAddressInPublish（獲取broker的網(wǎng)絡(luò)地址（主-master的地址）
public String findBrokerAddressInPublish(final String brokerName) {
    HashMap<Long/* brokerId */, String/* address */> map = this.brokerAddrTable.get(brokerName);
    if (map != null && !map.isEmpty()) {
        return map.get(MixAll.MASTER_ID);
    }
    return null;
}

備注：

brokerAddrTable 是路由更新維護的broker地址信息。

1.1.2 消息壓縮消息

int sysFlag = 0;
boolean msgBodyCompressed = false;//壓縮標記
if (this.tryToCompressMessage(msg)) {//嘗試壓縮
    sysFlag |= MessageSysFlag.COMPRESSED_FLAG;
    msgBodyCompressed = true;
}
// 壓縮
private boolean tryToCompressMessage(final Message msg) {
    if (msg instanceof MessageBatch) {
        //batch dose not support compressing right now
        return false;
    }
    byte[] body = msg.getBody();
    if (body != null) {
        if (body.length >= this.defaultMQProducer.getCompressMsgBodyOverHowmuch()) {
            try {
                byte[] data = UtilAll.compress(body, zipCompressLevel);
                if (data != null) {
                    msg.setBody(data);
                    return true;
                }
            } catch (IOException e) {
                log.error("tryToCompressMessage exception", e);
                log.warn(msg.toString());
            }
        }
    }
    return false;
}

備注:

  1. 批量消息不支持壓縮
  2. 消息大于4k，zip壓縮，壓縮級別：默認：5

1.1.3 發(fā)送消息請求參數(shù)構(gòu)建消息

  • SendMessageRequestHeader
/** 構(gòu)建消息發(fā)送 請求包 。主要包含如下重要信息:生產(chǎn)者組、主題名稱、默認創(chuàng)建主題Key、該主題在單個Broker默認隊列數(shù) 、隊列ID (隊列序號)、消息系統(tǒng)標記 ( MessageSysFlag)、
   消息發(fā)送時間、消息標記(RocketMQ對消息中的 flag不做任何處理， 供應(yīng)用程序使用)、 消息擴展屬性、消息重試次數(shù)、是否是批量消息等。
*/
SendMessageRequestHeader requestHeader = newSendMessageRequestHeader();
requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());//生產(chǎn)者組
requestHeader.setTopic(msg.getTopic());//主題名稱
requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());//默認創(chuàng)建主題Key
requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());//該主題在單個Broker默認隊列數(shù)
requestHeader.setQueueId(mq.getQueueId());//隊列ID (隊列序號)
requestHeader.setSysFlag(sysFlag);//消息系統(tǒng)標記 ( MessageSysFlag)
requestHeader.setBornTimestamp(System.currentTimeMillis());//消息發(fā)送時間
requestHeader.setFlag(msg.getFlag());//消息標記(RocketMQ對消息中的 flag不做任何處理， 供應(yīng)用程序使用)
requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));//【重要】消息擴展屬性
requestHeader.setReconsumeTimes(0);//消息重試次數(shù) 
requestHeader.setUnitMode(this.isUnitMode());
requestHeader.setBatch(msg instanceofMessageBatch);//是否是批量消息等
if(requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {//主題 topic 包含：RETRY
   String reconsumeTimes = MessageAccessor.getReconsumeTime(msg);
   if (reconsumeTimes != null) {
       requestHeader.setReconsumeTimes(Integer.valueOf(reconsumeTimes));
       MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_RECONSUME_TIME);
   }
   String maxReconsumeTimes = MessageAccessor.getMaxReconsumeTimes(msg);
   if (maxReconsumeTimes != null) {
       requestHeader.setMaxReconsumeTimes(Integer.valueOf(maxReconsumeTimes));
       MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_MAX_RECONSUME_TIMES);
   }
}

1.1.4 消息發(fā)送

  • MQClientAPIImpl#sendMessage

public SendResult sendMessage(
        final String addr,
        final String brokerName,
        final Message msg,
        final SendMessageRequestHeader requestHeader,
        final long timeoutMillis,
        final CommunicationMode communicationMode,
        final SendCallback sendCallback,
        final TopicPublishInfo topicPublishInfo,
        final MQClientInstance instance,
        final int retryTimesWhenSendFailed,
        final SendMessageContext context,
        final DefaultMQProducerImpl producer
    ) throws RemotingException, MQBrokerException, InterruptedException {
    long beginStartTime = System.currentTimeMillis();
    RemotingCommand request = null;
    if (sendSmartMsg || msg instanceof MessageBatch) {
        //默認smartMsg(智能) 或者 批量消息
        SendMessageRequestHeaderV2 requestHeaderV2 = SendMessageRequestHeaderV2.createSendMessageRequestHeaderV2(requestHeader);
        request = RemotingCommand.createRequestCommand(msg instanceof MessageBatch ? RequestCode.SEND_BATCH_MESSAGE : RequestCode.SEND_MESSAGE_V2, requestHeaderV2);
    } else {
        request = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, requestHeader);
    }
    request.setBody(msg.getBody());//設(shè)置消息內(nèi)容

    switch (communicationMode) {
        case ONEWAY://單向
            this.remotingClient.invokeOneway(addr, request, timeoutMillis);
            return null;
        case ASYNC://異步
            final AtomicInteger times = new AtomicInteger();
            long costTimeAsync = System.currentTimeMillis() - beginStartTime;
            if (timeoutMillis < costTimeAsync) {
                throw new RemotingTooMuchRequestException("sendMessage call timeout");
            }
            this.sendMessageAsync(addr, brokerName, msg, timeoutMillis - costTimeAsync, request, sendCallback, topicPublishInfo, instance,
                retryTimesWhenSendFailed, times, context, producer);
            return null;
        case SYNC://同步
            long costTimeSync = System.currentTimeMillis() - beginStartTime;
            if (timeoutMillis < costTimeSync) { //超時判斷
                throw new RemotingTooMuchRequestException("sendMessage call timeout");
            }
            return this.sendMessageSync(addr, brokerName, msg, timeoutMillis - costTimeSync, request);
        default:
            assert false;
            break;
    }
    return null;
}

分析：

從此處可知道，單向/異步/同步發(fā)送的實際差別了。單向發(fā)送直接返回null,同步需要等待返回結(jié)果，異步返回null但sendCallback會異步處理發(fā)送結(jié)果。
牛逼的你一定會去研究 invokeOneway、sendMessageAsync、sendMessageSync 三個方法的的源碼，其實很簡單。

二、結(jié)論

其實發(fā)送流程涉及代碼很多，這邊沒有一一分析，比如落下的一些可擴展的鉤子函數(shù)，netty網(wǎng)絡(luò)處理，最關(guān)鍵的是異常處理等，建議仔細研究哈。

程序員的核心競爭力其實還是技術(shù)，因此對技術(shù)還是要不斷的學(xué)習(xí)，關(guān)注 “IT 巔峰技術(shù)” 公眾號 ，該公眾號內(nèi)容定位：中高級開發(fā)、架構(gòu)師、中層管理人員等中高端崗位服務(wù)的，除了技術(shù)交流外還有很多架構(gòu)思想和實戰(zhàn)案例。

作者是 《 消息中間件 RocketMQ 技術(shù)內(nèi)幕》 一書作者，同時也是 “RocketMQ 上海社區(qū)”聯(lián)合創(chuàng)始人，曾就職于拼多多、德邦等公司，現(xiàn)任上市快遞公司架構(gòu)負責人，主要負責開發(fā)框架的搭建、中間件相關(guān)技術(shù)的二次開發(fā)和運維管理、混合云及基礎(chǔ)服務(wù)平臺的建設(shè)。

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容