前 言
在深入講解消息發(fā)送之前疗涉，我們可先簡(jiǎn)單概括消息的發(fā)送的主要步驟可分為：消息驗(yàn)證、路由查詢吟秩、選擇消息隊(duì)列咱扣、消息組裝、消息發(fā)送涵防、消息結(jié)果處理闹伪、異常處理；(單向發(fā)送并不處理消息發(fā)送結(jié)果)；同步偏瓤、異步杀怠、單向發(fā)送消息的入口API有一些區(qū)別，本文將以下面接口實(shí)現(xiàn)類為入口分析消息發(fā)送的流程：
DefaultMQProducerImpl#sendDefaultImpl
(由于消息發(fā)送細(xì)節(jié)非常多厅克，本文將分析核心步驟赔退，如漏掉還請(qǐng)各位查漏補(bǔ)缺，自行分析哈)
同步發(fā)送總結(jié)流程圖如下：
image.png
一证舟、源碼分析
DefaultMQProducerImpl#sendDefaultImpl
/**
* 發(fā)送信息
* @param msg 消息內(nèi)容
* @param communicationMode 發(fā)送模式
* @param sendCallback 回掉
* @param timeout 超時(shí)時(shí)間
*/
private SendResult sendDefaultImpl(
Message msg,
final CommunicationMode communicationMode,
final SendCallback sendCallback,
final long timeout
) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
this.makeSureStateOK(); //驗(yàn)證 serviceState == Running 運(yùn)行中
Validators.checkMessage(msg, this.defaultMQProducer); //1> 驗(yàn)證消息
final long invokeID = random.nextLong();//隨機(jī)的-invokeId
long beginTimestampFirst = System.currentTimeMillis();//開始時(shí)間
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ù)硕旗，同步默認(rèn)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>選擇消息隊(duì)列
if (mqSelected != null) {
mq = mqSelected;
brokersSent[times] = mq.getBrokerName();
try {
beginTimestampPrev = System.currentTimeMillis();//本次開始時(shí)間
long costTime = beginTimestampPrev - beginTimestampFirst;//計(jì)算發(fā)送消耗時(shí)間
if (timeout < costTime) {//如果消耗時(shí)間 大于 超時(shí)時(shí)間褪储，直接break
callTimeout = true;
break;
}
//發(fā)送消息
sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime); // 4>消息發(fā)送
//發(fā)送完成時(shí)間
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 { //沒有找到消息隊(duì)列卵渴，直接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 驗(yàn)證消息
Validators.checkMessage
//Validate message 驗(yàn)證消息
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()); // 驗(yàn)證topic, 此處代碼大家可自行查看，灰常簡(jiǎn)單
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");
}
//消息最大長(zhǎng)度 不能大于 4M
if (msg.getBody().length > defaultMQProducer.getMaxMessageSize()) {
throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL,
"the message body size over max value, MAX: " + defaultMQProducer.getMaxMessageSize());
}
}
備注:
主要驗(yàn)證消息分為兩部分
- topic驗(yàn)證:匹配正則表達(dá)式(^[%|a-zA-Z0-9_-]+$),長(zhǎng)度小于255,不等于默認(rèn)主題:TBW102
- body驗(yàn)證:body內(nèi)容是否為空鲤竹，消息內(nèi)容最大長(zhǎng)度默認(rèn)不能超過4M
1.2 獲取路由信息
tryToFindTopicPublishInfo
在'路由動(dòng)態(tài)更新'我們以及分析過了，代碼大家可以再回顧下昔榴，簡(jiǎn)單邏輯總結(jié)如下:
- 如果生產(chǎn)者中緩存了 topic 的路由信息辛藻，如果該路由信息中包含了消息隊(duì)列，則直接返回該路由信息;
- 如果沒有緩存或沒有包含消息隊(duì)列互订， 則向 NameServer查詢?cè)?topic 的路由信息;
- 如果最終未找到路由信息吱肌，則拋出異常 : 無法找到主題相關(guān)路由信息異常.
1.3 選擇消息隊(duì)列
將在'系列5'著重分析此段代碼功能消息
1.4 消息發(fā)送
sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
由于代碼篇幅太長(zhǎng),下面講解只摘取sendKernelImpl方法的核心代碼解析,但強(qiáng)烈建議仔細(xì)去擼一遍代碼消息。
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 是路由更新維護(hù)的broker地址信息氮墨。
1.1.2 消息壓縮消息
int sysFlag = 0;
boolean msgBodyCompressed = false;//壓縮標(biāo)記
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;
}
備注:
- 批量消息不支持壓縮
- 消息大于4k，zip壓縮吐葵，壓縮級(jí)別：默認(rèn)：5
1.1.3 發(fā)送消息請(qǐng)求參數(shù)構(gòu)建消息
- SendMessageRequestHeader
/** 構(gòu)建消息發(fā)送 請(qǐng)求包 规揪。主要包含如下重要信息:生產(chǎn)者組、主題名稱温峭、默認(rèn)創(chuàng)建主題Key猛铅、該主題在單個(gè)Broker默認(rèn)隊(duì)列數(shù) 、隊(duì)列ID (隊(duì)列序號(hào))凤藏、消息系統(tǒng)標(biāo)記 ( MessageSysFlag)奸忽、
消息發(fā)送時(shí)間、消息標(biāo)記(RocketMQ對(duì)消息中的 flag不做任何處理揖庄， 供應(yīng)用程序使用)栗菜、 消息擴(kuò)展屬性、消息重試次數(shù)蹄梢、是否是批量消息等疙筹。
*/
SendMessageRequestHeader requestHeader = newSendMessageRequestHeader();
requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());//生產(chǎn)者組
requestHeader.setTopic(msg.getTopic());//主題名稱
requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());//默認(rèn)創(chuàng)建主題Key
requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());//該主題在單個(gè)Broker默認(rèn)隊(duì)列數(shù)
requestHeader.setQueueId(mq.getQueueId());//隊(duì)列ID (隊(duì)列序號(hào))
requestHeader.setSysFlag(sysFlag);//消息系統(tǒng)標(biāo)記 ( MessageSysFlag)
requestHeader.setBornTimestamp(System.currentTimeMillis());//消息發(fā)送時(shí)間
requestHeader.setFlag(msg.getFlag());//消息標(biāo)記(RocketMQ對(duì)消息中的 flag不做任何處理， 供應(yīng)用程序使用)
requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));//【重要】消息擴(kuò)展屬性
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) {
//默認(rèn)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) { //超時(shí)判斷
throw new RemotingTooMuchRequestException("sendMessage call timeout");
}
return this.sendMessageSync(addr, brokerName, msg, timeoutMillis - costTimeSync, request);
default:
assert false;
break;
}
return null;
}
分析：
從此處可知道，單向/異步/同步發(fā)送的實(shí)際差別了腌歉。單向發(fā)送直接返回null,同步需要等待返回結(jié)果蛙酪，異步返回null但sendCallback會(huì)異步處理發(fā)送結(jié)果。
牛逼的你一定會(huì)去研究 invokeOneway翘盖、sendMessageAsync桂塞、sendMessageSync 三個(gè)方法的的源碼，其實(shí)很簡(jiǎn)單馍驯。
二阁危、結(jié)論
其實(shí)發(fā)送流程涉及代碼很多，這邊沒有一一分析汰瘫，比如落下的一些可擴(kuò)展的鉤子函數(shù)狂打，netty網(wǎng)絡(luò)處理，最關(guān)鍵的是異常處理等混弥，建議仔細(xì)研究哈趴乡。
程序員的核心競(jìng)爭(zhēng)力其實(shí)還是技術(shù)，因此對(duì)技術(shù)還是要不斷的學(xué)習(xí)蝗拿，關(guān)注 “IT 巔峰技術(shù)” 公眾號(hào) 晾捏，該公眾號(hào)內(nèi)容定位：中高級(jí)開發(fā)、架構(gòu)師哀托、中層管理人員等中高端崗位服務(wù)的惦辛，除了技術(shù)交流外還有很多架構(gòu)思想和實(shí)戰(zhàn)案例。
作者是 《 消息中間件 RocketMQ 技術(shù)內(nèi)幕》 一書作者仓手，同時(shí)也是 “RocketMQ 上海社區(qū)”聯(lián)合創(chuàng)始人胖齐，曾就職于拼多多、德邦等公司嗽冒，現(xiàn)任上市快遞公司架構(gòu)負(fù)責(zé)人呀伙，主要負(fù)責(zé)開發(fā)框架的搭建、中間件相關(guān)技術(shù)的二次開發(fā)和運(yùn)維管理辛慰、混合云及基礎(chǔ)服務(wù)平臺(tái)的建設(shè)区匠。
