溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務(wù)條款》
您好，登錄后才能下訂單哦！
這篇文章主要講解了“RocketMQ消息軌跡是怎樣的”，文中的講解內(nèi)容簡單清晰，易于學(xué)習(xí)與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學(xué)習(xí)“RocketMQ消息軌跡是怎樣的”吧！
首先我們來看一下在消息發(fā)送端如何啟用消息軌跡，示例代碼如下：
public class TraceProducer {
public static void main(String[] args) throws MQClientException, InterruptedException {
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName",true); // [@1](https://my.oschina.net/u/1198)
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();
for (int i = 0; i < 10; i++)
try {
{
Message msg = new Message("TopicTest",
"TagA",
"OrderID188",
"Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
SendResult sendResult = producer.send(msg);
System.out.printf("%s%n", sendResult);
}
} catch (Exception e) {
e.printStackTrace();
}
producer.shutdown();
}
}從上述代碼可以看出其關(guān)鍵點是在創(chuàng)建DefaultMQProducer時指定開啟消息軌跡跟蹤。我們不妨瀏覽一下DefaultMQProducer與啟用消息軌跡相關(guān)的構(gòu)造函數(shù)：
public DefaultMQProducer(final String producerGroup, boolean enableMsgTrace) public DefaultMQProducer(final String producerGroup, boolean enableMsgTrace, final String customizedTraceTopic)
參數(shù)如下：
String producerGroup 生產(chǎn)者所屬組名。
boolean enableMsgTrace 是否開啟跟蹤消息軌跡，默認(rèn)為false。
String customizedTraceTopic 如果開啟消息軌跡跟蹤，用來存儲消息軌跡數(shù)據(jù)所屬的主題名稱，默認(rèn)為：RMQ_SYS_TRACE_TOPIC。
public DefaultMQProducer(final String producerGroup, RPCHook rpcHook, boolean enableMsgTrace,final String customizedTraceTopic) { // [@1](https://my.oschina.net/u/1198)
this.producerGroup = producerGroup;
defaultMQProducerImpl = new DefaultMQProducerImpl(this, rpcHook);
//if client open the message trace feature
if (enableMsgTrace) { // @2
try {
AsyncTraceDispatcher dispatcher = new AsyncTraceDispatcher(customizedTraceTopic, rpcHook);
dispatcher.setHostProducer(this.getDefaultMQProducerImpl());
traceDispatcher = dispatcher;
this.getDefaultMQProducerImpl().registerSendMessageHook(
new SendMessageTraceHookImpl(traceDispatcher)); // [@3](https://my.oschina.net/u/2648711)
} catch (Throwable e) {
log.error("system mqtrace hook init failed ,maybe can't send msg trace data");
}
}
}代碼@1：首先介紹一下其局部變量。
String producerGroup 生產(chǎn)者所屬組。
RPCHook rpcHook 生產(chǎn)者發(fā)送鉤子函數(shù)。
boolean enableMsgTrace 是否開啟消息軌跡跟蹤。
String customizedTraceTopic 定制用于存儲消息軌跡的數(shù)據(jù)。
代碼@2：用來構(gòu)建AsyncTraceDispatcher，看其名：異步轉(zhuǎn)發(fā)消息軌跡數(shù)據(jù)，稍后重點關(guān)注。
代碼@3：構(gòu)建SendMessageTraceHookImpl對象，并使用AsyncTraceDispatcher用來異步轉(zhuǎn)發(fā)。
SendMessageHook 消息發(fā)送鉤子函數(shù)，用于在消息發(fā)送之前、發(fā)送之后執(zhí)行一定的業(yè)務(wù)邏輯，是記錄消息軌跡的最佳擴展點。
TraceDispatcher 消息軌跡轉(zhuǎn)發(fā)處理器，其默認(rèn)實現(xiàn)類AsyncTraceDispatcher，異步實現(xiàn)消息軌跡數(shù)據(jù)的發(fā)送。下面對其屬性做一個簡單的介紹：
int queueSize 異步轉(zhuǎn)發(fā)，隊列長度，默認(rèn)為2048，當(dāng)前版本不能修改。
int batchSize 批量消息條數(shù)，消息軌跡一次消息發(fā)送請求包含的數(shù)據(jù)條數(shù)，默認(rèn)為100，當(dāng)前版本不能修改。
int maxMsgSize 消息軌跡一次發(fā)送的最大消息大小，默認(rèn)為128K，當(dāng)前版本不能修改。
DefaultMQProducer traceProducer 用來發(fā)送消息軌跡的消息發(fā)送者。
ThreadPoolExecutor traceExecuter 線程池，用來異步執(zhí)行消息發(fā)送。
AtomicLong discardCount 記錄丟棄的消息個數(shù)。
Thread worker woker線程，主要負(fù)責(zé)從追加隊列中獲取一批待發(fā)送的消息軌跡數(shù)據(jù)，提交到線程池中執(zhí)行。
ArrayBlockingQueue< TraceContext> traceContextQueue 消息軌跡TraceContext隊列，用來存放待發(fā)送到服務(wù)端的消息。
ArrayBlockingQueue< Runnable> appenderQueue 線程池內(nèi)部隊列，默認(rèn)長度1024。
DefaultMQPushConsumerImpl hostConsumer 消費者信息，記錄消息消費時的軌跡信息。
String traceTopicName 用于跟蹤消息軌跡的topic名稱。
public void sendMessageBefore(SendMessageContext context) {
//if it is message trace data,then it doesn't recorded
if (context == null || context.getMessage().getTopic().startsWith(((AsyncTraceDispatcher) localDispatcher).getTraceTopicName())) { // @1
return;
}
//build the context content of TuxeTraceContext
TraceContext tuxeContext = new TraceContext();
tuxeContext.setTraceBeans(new ArrayList<tracebean>(1));
context.setMqTraceContext(tuxeContext);
tuxeContext.setTraceType(TraceType.Pub);
tuxeContext.setGroupName(context.getProducerGroup()); // @2
//build the data bean object of message trace
TraceBean traceBean = new TraceBean(); // @3
traceBean.setTopic(context.getMessage().getTopic());
traceBean.setTags(context.getMessage().getTags());
traceBean.setKeys(context.getMessage().getKeys());
traceBean.setStoreHost(context.getBrokerAddr());
traceBean.setBodyLength(context.getMessage().getBody().length);
traceBean.setMsgType(context.getMsgType());
tuxeContext.getTraceBeans().add(traceBean);
}代碼@1：如果topic主題為消息軌跡的Topic，直接返回。
代碼@2：在消息發(fā)送上下文中，設(shè)置用來跟蹤消息軌跡的上下環(huán)境，里面主要包含一個TraceBean集合、追蹤類型（TraceType.Pub）與生產(chǎn)者所屬的組。
代碼@3：構(gòu)建一條跟蹤消息，用TraceBean來表示，記錄原消息的topic、tags、keys、發(fā)送到broker地址、消息體長度等消息。
從上文看出，sendMessageBefore主要的用途就是在消息發(fā)送的時候，先準(zhǔn)備一部分消息跟蹤日志，存儲在發(fā)送上下文環(huán)境中，此時并不會發(fā)送消息軌跡數(shù)據(jù)。
public void sendMessageAfter(SendMessageContext context) {
//if it is message trace data,then it doesn't recorded
if (context == null || context.getMessage().getTopic().startsWith(((AsyncTraceDispatcher) localDispatcher).getTraceTopicName()) // @1
|| context.getMqTraceContext() == null) {
return;
}
if (context.getSendResult() == null) {
return;
}
if (context.getSendResult().getRegionId() == null
|| !context.getSendResult().isTraceOn()) {
// if switch is false,skip it
return;
}
TraceContext tuxeContext = (TraceContext) context.getMqTraceContext();
TraceBean traceBean = tuxeContext.getTraceBeans().get(0); // @2
int costTime = (int) ((System.currentTimeMillis() - tuxeContext.getTimeStamp()) / tuxeContext.getTraceBeans().size()); // @3
tuxeContext.setCostTime(costTime); // @4
if (context.getSendResult().getSendStatus().equals(SendStatus.SEND_OK)) {
tuxeContext.setSuccess(true);
} else {
tuxeContext.setSuccess(false);
}
tuxeContext.setRegionId(context.getSendResult().getRegionId());
traceBean.setMsgId(context.getSendResult().getMsgId());
traceBean.setOffsetMsgId(context.getSendResult().getOffsetMsgId());
traceBean.setStoreTime(tuxeContext.getTimeStamp() + costTime / 2);
localDispatcher.append(tuxeContext); // @5
}代碼@1：如果topic主題為消息軌跡的Topic，直接返回。
代碼@2：從MqTraceContext中獲取跟蹤的TraceBean，雖然設(shè)計成List結(jié)構(gòu)體，但在消息發(fā)送場景，這里的數(shù)據(jù)永遠(yuǎn)只有一條，及時是批量發(fā)送也不例外。
代碼@3：獲取消息發(fā)送到收到響應(yīng)結(jié)果的耗時。
代碼@4：設(shè)置costTime(耗時)、success(是否發(fā)送成功)、regionId(發(fā)送到broker所在的分區(qū))、msgId(消息ID，全局唯一)、offsetMsgId(消息物理偏移量，如果是批量消息，則是最后一條消息的物理偏移量)、storeTime，這里使用的是(客戶端發(fā)送時間 + 二分之一的耗時)來表示消息的存儲時間，這里是一個估值。
代碼@5：將需要跟蹤的信息通過TraceDispatcher轉(zhuǎn)發(fā)到Broker服務(wù)器。其代碼如下：
public boolean append(final Object ctx) {
boolean result = traceContextQueue.offer((TraceContext) ctx);
if (!result) {
log.info("buffer full" + discardCount.incrementAndGet() + " ,context is " + ctx);
}
return result;
}這里一個非常關(guān)鍵的點是offer方法的使用，當(dāng)隊列無法容納新的元素時會立即返回false，并不會阻塞。
接下來將目光轉(zhuǎn)向TraceDispatcher的實現(xiàn)。
TraceDispatcher，用于客戶端消息軌跡數(shù)據(jù)轉(zhuǎn)發(fā)到Broker，其默認(rèn)實現(xiàn)類：AsyncTraceDispatcher。
public AsyncTraceDispatcher(String traceTopicName, RPCHook rpcHook) throws MQClientException {
// queueSize is greater than or equal to the n power of 2 of value
this.queueSize = 2048;
this.batchSize = 100;
this.maxMsgSize = 128000;
this.discardCount = new AtomicLong(0L);
this.traceContextQueue = new ArrayBlockingQueue<tracecontext>(1024);
this.appenderQueue = new ArrayBlockingQueue<runnable>(queueSize);
if (!UtilAll.isBlank(traceTopicName)) {
this.traceTopicName = traceTopicName;
} else {
this.traceTopicName = MixAll.RMQ_SYS_TRACE_TOPIC;
} // @1
this.traceExecuter = new ThreadPoolExecutor(// :
10, //
20, //
1000 * 60, //
TimeUnit.MILLISECONDS, //
this.appenderQueue, //
new ThreadFactoryImpl("MQTraceSendThread_"));
traceProducer = getAndCreateTraceProducer(rpcHook); // @2
}代碼@1：初始化核心屬性，該版本這些值都是“固化”的，用戶無法修改。
queueSize 隊列長度，默認(rèn)為2048，異步線程池能夠積壓的消息軌跡數(shù)量。
batchSize 一次向Broker批量發(fā)送的消息條數(shù)，默認(rèn)為100.
maxMsgSize 向Broker匯報消息軌跡時，消息體的總大小不能超過該值，默認(rèn)為128k。
discardCount 整個運行過程中，丟棄的消息軌跡數(shù)據(jù)，這里要說明一點的是，如果消息TPS發(fā)送過大，異步轉(zhuǎn)發(fā)線程處理不過來時，會主動丟棄消息軌跡數(shù)據(jù)。
traceContextQueue traceContext積壓隊列，客戶端(消息發(fā)送、消息消費者)在收到處理結(jié)果后，將消息軌跡提交到噶隊列中，則會立即返回。
appenderQueue 提交到Broker線程池中隊列。
traceTopicName 用于接收消息軌跡的Topic，默認(rèn)為RMQ_SYS_TRANS_HALF_TOPIC。
traceExecuter 用于發(fā)送到Broker服務(wù)的異步線程池，核心線程數(shù)默認(rèn)為10，最大線程池為20，隊列堆積長度2048，線程名稱：MQTraceSendThread_。、
traceProducer 發(fā)送消息軌跡的Producer。
代碼@2：調(diào)用getAndCreateTraceProducer方法創(chuàng)建用于發(fā)送消息軌跡的Producer(消息發(fā)送者)，下面詳細(xì)介紹一下其實現(xiàn)。
private DefaultMQProducer getAndCreateTraceProducer(RPCHook rpcHook) {
DefaultMQProducer traceProducerInstance = this.traceProducer;
if (traceProducerInstance == null) { //@1
traceProducerInstance = new DefaultMQProducer(rpcHook);
traceProducerInstance.setProducerGroup(TraceConstants.GROUP_NAME);
traceProducerInstance.setSendMsgTimeout(5000);
traceProducerInstance.setVipChannelEnabled(false);
// The max size of message is 128K
traceProducerInstance.setMaxMessageSize(maxMsgSize - 10 * 1000);
}
return traceProducerInstance;
}代碼@1：如果還未建立發(fā)送者，則創(chuàng)建用于發(fā)送消息軌跡的消息發(fā)送者，其GroupName為：_INNER_TRACE_PRODUCER，消息發(fā)送超時時間5s，最大允許發(fā)送消息大小118K。
public void start(String nameSrvAddr) throws MQClientException {
if (isStarted.compareAndSet(false, true)) { // @1
traceProducer.setNamesrvAddr(nameSrvAddr);
traceProducer.setInstanceName(TRACE_INSTANCE_NAME + "_" + nameSrvAddr);
traceProducer.start();
}
this.worker = new Thread(new AsyncRunnable(), "MQ-AsyncTraceDispatcher-Thread-" + dispatcherId); // @2
this.worker.setDaemon(true);
this.worker.start();
this.registerShutDownHook();
}開始啟動，其調(diào)用的時機為啟動DefaultMQProducer時，如果啟用跟蹤消息軌跡，則調(diào)用之。
代碼@1：如果用于發(fā)送消息軌跡的發(fā)送者沒有啟動，則設(shè)置nameserver地址，并啟動著。
代碼@2：啟動一個線程，用于執(zhí)行AsyncRunnable任務(wù)，接下來將重點介紹。
class AsyncRunnable implements Runnable {
private boolean stopped;
public void run() {
while (!stopped) {
List<tracecontext> contexts = new ArrayList<tracecontext>(batchSize); // @1
for (int i = 0; i < batchSize; i++) {
TraceContext context = null;
try {
//get trace data element from blocking Queue — traceContextQueue
context = traceContextQueue.poll(5, TimeUnit.MILLISECONDS); // @2
} catch (InterruptedException e) {
}
if (context != null) {
contexts.add(context);
} else {
break;
}
}
if (contexts.size() > 0) { :
AsyncAppenderRequest request = new AsyncAppenderRequest(contexts); // @3
traceExecuter.submit(request);
} else if (AsyncTraceDispatcher.this.stopped) {
this.stopped = true;
}
}
}
}代碼@1：構(gòu)建待提交消息跟蹤Bean，每次最多發(fā)送batchSize，默認(rèn)為100條。
代碼@2：從traceContextQueue中取出一個待提交的TraceContext，設(shè)置超時時間為5s，即如何該隊列中沒有待提交的TraceContext，則最多等待5s。
代碼@3：向線程池中提交任務(wù)AsyncAppenderRequest。
public void sendTraceData(List<tracecontext> contextList) {
Map<string, list<tracetransferbean>> transBeanMap = new HashMap<string, list<tracetransferbean>>();
for (TraceContext context : contextList) { //@1
if (context.getTraceBeans().isEmpty()) {
continue;
}
// Topic value corresponding to original message entity content
String topic = context.getTraceBeans().get(0).getTopic(); // @2
// Use original message entity's topic as key
String key = topic;
List<tracetransferbean> transBeanList = transBeanMap.get(key);
if (transBeanList == null) {
transBeanList = new ArrayList<tracetransferbean>();
transBeanMap.put(key, transBeanList);
}
TraceTransferBean traceData = TraceDataEncoder.encoderFromContextBean(context); // @3
transBeanList.add(traceData);
}
for (Map.Entry<string, list<tracetransferbean>> entry : transBeanMap.entrySet()) { // @4
flushData(entry.getValue());
}
}代碼@1：遍歷收集的消息軌跡數(shù)據(jù)。
代碼@2：獲取存儲消息軌跡的Topic。
代碼@3：對TraceContext進行編碼，這里是消息軌跡的傳輸數(shù)據(jù)，稍后對其詳細(xì)看一下，了解其上傳的格式。
代碼@4：將編碼后的數(shù)據(jù)發(fā)送到Broker服務(wù)器。
根據(jù)消息軌跡跟蹤類型，其格式會有一些不一樣，下面分別來介紹其合適。
case Pub: {
TraceBean bean = ctx.getTraceBeans().get(0);
//append the content of context and traceBean to transferBean's TransData
sb.append(ctx.getTraceType()).append(TraceConstants.CONTENT_SPLITOR)//
.append(ctx.getTimeStamp()).append(TraceConstants.CONTENT_SPLITOR)//
.append(ctx.getRegionId()).append(TraceConstants.CONTENT_SPLITOR)//
.append(ctx.getGroupName()).append(TraceConstants.CONTENT_SPLITOR)//
.append(bean.getTopic()).append(TraceConstants.CONTENT_SPLITOR)//
.append(bean.getMsgId()).append(TraceConstants.CONTENT_SPLITOR)//
.append(bean.getTags()).append(TraceConstants.CONTENT_SPLITOR)//
.append(bean.getKeys()).append(TraceConstants.CONTENT_SPLITOR)//
.append(bean.getStoreHost()).append(TraceConstants.CONTENT_SPLITOR)//
.append(bean.getBodyLength()).append(TraceConstants.CONTENT_SPLITOR)//
.append(ctx.getCostTime()).append(TraceConstants.CONTENT_SPLITOR)//
.append(bean.getMsgType().ordinal()).append(TraceConstants.CONTENT_SPLITOR)//
.append(bean.getOffsetMsgId()).append(TraceConstants.CONTENT_SPLITOR)//
.append(ctx.isSuccess()).append(TraceConstants.FIELD_SPLITOR);
}消息軌跡數(shù)據(jù)的協(xié)議使用字符串拼接，字段的分隔符號為1，整個數(shù)據(jù)以2結(jié)尾，感覺這個設(shè)計還是有點“不可思議”，為什么不直接使用json協(xié)議呢？
for (TraceBean bean : ctx.getTraceBeans()) {
sb.append(ctx.getTraceType()).append(TraceConstants.CONTENT_SPLITOR)//
.append(ctx.getTimeStamp()).append(TraceConstants.CONTENT_SPLITOR)//
.append(ctx.getRegionId()).append(TraceConstants.CONTENT_SPLITOR)//
.append(ctx.getGroupName()).append(TraceConstants.CONTENT_SPLITOR)//
.append(ctx.getRequestId()).append(TraceConstants.CONTENT_SPLITOR)//
.append(bean.getMsgId()).append(TraceConstants.CONTENT_SPLITOR)//
.append(bean.getRetryTimes()).append(TraceConstants.CONTENT_SPLITOR)//
.append(bean.getKeys()).append(TraceConstants.FIELD_SPLITOR);//
}
}軌跡就是按照上述順序拼接而成，各個字段使用1分隔，每一條記錄使用2結(jié)尾。
case SubAfter: {
for (TraceBean bean : ctx.getTraceBeans()) {
sb.append(ctx.getTraceType()).append(TraceConstants.CONTENT_SPLITOR)//
.append(ctx.getRequestId()).append(TraceConstants.CONTENT_SPLITOR)//
.append(bean.getMsgId()).append(TraceConstants.CONTENT_SPLITOR)//
.append(ctx.getCostTime()).append(TraceConstants.CONTENT_SPLITOR)//
.append(ctx.isSuccess()).append(TraceConstants.CONTENT_SPLITOR)//
.append(bean.getKeys()).append(TraceConstants.CONTENT_SPLITOR)//
.append(ctx.getContextCode()).append(TraceConstants.FIELD_SPLITOR);
}
}
}格式編碼一樣，就不重復(fù)多說。
經(jīng)過上面的源碼跟蹤，消息發(fā)送端的消息軌跡跟蹤流程、消息軌跡數(shù)據(jù)編碼協(xié)議就清晰了，接下來我們使用一張序列圖來結(jié)束本部分的講解。
其實行文至此，只關(guān)注了消息發(fā)送的消息軌跡跟蹤，消息消費的軌跡跟蹤又是如何呢？其實現(xiàn)原理其實是一樣的，就是在消息消費前后執(zhí)行特定的鉤子函數(shù)，其實現(xiàn)類為ConsumeMessageTraceHookImpl，由于其實現(xiàn)與消息發(fā)送的思路類似，故就不詳細(xì)介紹了。
其實從上面的分析，我們已經(jīng)得知，RocketMQ的消息軌跡數(shù)據(jù)存儲在到Broker上，那消息軌跡的主題名如何指定？其路由信息又怎么分配才好呢？是每臺Broker上都創(chuàng)建還是只在其中某臺上創(chuàng)建呢？RocketMQ支持系統(tǒng)默認(rèn)與自定義消息軌跡的主題。
RocketMQ默認(rèn)的消息軌跡主題為：RMQ_SYS_TRACE_TOPIC，那該Topic需要手工創(chuàng)建嗎？其路由信息呢？
{
if (this.brokerController.getBrokerConfig().isTraceTopicEnable()) { // @1
String topic = this.brokerController.getBrokerConfig().getMsgTraceTopicName();
TopicConfig topicConfig = new TopicConfig(topic);
this.systemTopicList.add(topic);
topicConfig.setReadQueueNums(1); // @2
topicConfig.setWriteQueueNums(1);
this.topicConfigTable.put(topicConfig.getTopicName(), topicConfig);
}
}上述代碼出自TopicConfigManager的構(gòu)造函數(shù)，在Broker啟動的時候會創(chuàng)建topicConfigManager對象，用來管理topic的路由信息。
代碼@1：如果Broker開啟了消息軌跡跟蹤(traceTopicEnable=true)時，會自動創(chuàng)建默認(rèn)消息軌跡的topic路由信息，注意其讀寫隊列數(shù)為1。
在創(chuàng)建消息發(fā)送者、消息消費者時，可以顯示的指定消息軌跡的Topic，例如：
public DefaultMQProducer(final String producerGroup, RPCHook rpcHook, boolean enableMsgTrace,final String customizedTraceTopic) public DefaultMQPushConsumer(final String consumerGroup, RPCHook rpcHook, AllocateMessageQueueStrategy allocateMessageQueueStrategy, boolean enableMsgTrace, final String customizedTraceTopic)
通過customizedTraceTopic來指定消息軌跡Topic。
溫馨提示：通常在生產(chǎn)環(huán)境上，將不會開啟自動創(chuàng)建主題，故需要RocketMQ運維管理人員提前創(chuàng)建好Topic。
感謝各位的閱讀，以上就是“RocketMQ消息軌跡是怎樣的”的內(nèi)容了，經(jīng)過本文的學(xué)習(xí)后，相信大家對RocketMQ消息軌跡是怎樣的這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關(guān)知識點的文章，歡迎關(guān)注！
免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點不代表本網(wǎng)站立場，如果涉及侵權(quán)請聯(lián)系站長郵箱：is@yisu.com進行舉報，并提供相關(guān)證據(jù)，一經(jīng)查實，將立刻刪除涉嫌侵權(quán)內(nèi)容。
億速云公眾號
手機網(wǎng)站二維碼