前言

得益于MQ削峰填谷，系統(tǒng)解耦，操作異步等功能特性，在互聯(lián)網(wǎng)行業(yè)，可以說有分布式服務(wù)的地方，MQ都往往不會缺席。由阿里自研的RocketMQ更是經(jīng)歷了多年的雙十一高并發(fā)挑戰(zhàn)，其中4.3.0版本推出了事務(wù)消息的新特性，本文對RocketMQ 4.5.0版本事務(wù)消息相關(guān)的源碼跟蹤介紹，通過閱讀讀者可以知道：

事務(wù)消息解決什么樣的問題

事務(wù)消息的實(shí)現(xiàn)原理及其設(shè)計(jì)亮點(diǎn)

解決什么問題

假設(shè)我所在的系統(tǒng)現(xiàn)在有這樣一個(gè)場景：

本地開啟數(shù)據(jù)庫事務(wù)進(jìn)行扣款操作，成功后發(fā)送MQ消息給庫存中心進(jìn)行發(fā)貨。

有人會想到開啟mybatis事務(wù)實(shí)現(xiàn)，把本地事務(wù)和MQ消息放在一起不就行了嗎？如果MQ發(fā)送成功，就提交事務(wù)，發(fā)送失敗就回滾事務(wù)，整套操作一氣呵成。

transaction{? 扣款();booleansuccess = 發(fā)送MQ();if(success){? ? commit();? }else{? ? rollBack();? }}

看似沒什么問題，但是網(wǎng)絡(luò)是不可靠的。

假設(shè)MQ返回過來的響應(yīng)因?yàn)榫W(wǎng)絡(luò)原因遲遲沒有收到，所以在面對不確定的MQ返回結(jié)果只好進(jìn)行回滾。但是MQ 服務(wù)器又確實(shí)是收到了這條消息的，只是給客戶端的響應(yīng)丟失了，所以導(dǎo)致的結(jié)果就是扣款失敗，成功發(fā)貨。

既然MQ消息的發(fā)送不能和本地事務(wù)寫在一起，那如何來保證其整體具有原子性的需求呢？答案就是今天我們介紹的主角：事務(wù)消息。

概覽

總體而言RocketMQ事務(wù)消息分為兩條主線

定時(shí)任務(wù)發(fā)送流程：發(fā)送half message(半消息)，執(zhí)行本地事務(wù)，發(fā)送事務(wù)執(zhí)行結(jié)果

定時(shí)任務(wù)回查流程：MQ服務(wù)器回查本地事務(wù)，發(fā)送事務(wù)執(zhí)行結(jié)果

因此本文也通過這兩條主線對源碼進(jìn)行分析

源碼分析

半消息發(fā)送流程

本地應(yīng)用(client)

在本地應(yīng)用發(fā)送事務(wù)消息的核心類是TransactionMQProducer，該類通過繼承DefaultMQProducer來復(fù)用大部分發(fā)送消息相關(guān)的邏輯，這個(gè)類的代碼量非常少只有100來行，下面是這個(gè)類的sendMessageTransaction方法

@OverridepublicTransactionSendResultsendMessageInTransaction(finalMessage msg,finalObject arg)throwsMQClientException{if(null==this.transactionListener) {thrownewMQClientException("TransactionListener is null",null);? ? }returnthis.defaultMQProducerImpl.sendMessageInTransaction(msg,null, arg);}

這個(gè)方法做了兩件事，

檢查transactionListener是否存在

調(diào)用父類執(zhí)行事務(wù)消息發(fā)送

TransactionListener在事務(wù)消息流程中起到至關(guān)重要的作用，一起看看這個(gè)接口

publicinterfaceTransactionListener{/**? ? * When send transactional prepare(half) message succeed, this method will be invoked to execute local transaction.? ? *? ? *@parammsg Half(prepare) message? ? *@paramarg Custom business parameter? ? *@returnTransaction state? ? */LocalTransactionStateexecuteLocalTransaction(finalMessage msg,finalObject arg);/**? ? * When no response to prepare(half) message. broker will send check message to check the transaction status, and this? ? * method will be invoked to get local transaction status.? ? *? ? *@parammsg Check message? ? *@returnTransaction state? ? */LocalTransactionStatecheckLocalTransaction(finalMessageExt msg);}

接口注釋說的很明白，配合上面的概覽圖來看就是，executeLocalTransaction方法對應(yīng)的就是執(zhí)行本地事務(wù)操作，checkLocalTransaction對應(yīng)的就是回查本地事務(wù)操作。

下面是DefaultMQProducer類的sendMessageInTransaction方法源碼

publicTransactionSendResultsendMessageInTransaction(finalMessage msg,finalLocalTransactionExecuter localTransactionExecuter,finalObject arg)throwsMQClientException{? ? ...? ? SendResult sendResult =null;? ? MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED,"true");? ? MessageAccessor.putProperty(msg, MessageConst.PROPERTY_PRODUCER_GROUP,this.defaultMQProducer.getProducerGroup());? ? ...? ? ? ? sendResult =this.send(msg);? ? ...switch(sendResult.getSendStatus()) {caseSEND_OK: {? ? ? ? ? ? ...? ? ? ? localTransactionState = transactionListener.executeLocalTransaction(msg, arg);? ? ? ? ? ? ? ? ...break;caseFLUSH_DISK_TIMEOUT:caseFLUSH_SLAVE_TIMEOUT:caseSLAVE_NOT_AVAILABLE:? ? ? ? ? ? localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;break;default:break;? ? }? ? ...this.endTransaction(sendResult, localTransactionState, localException);...}復(fù)制代碼

為了使源碼的邏輯更加直觀，筆者精簡了核心代碼。sendMessageInTransaction方法主要做了以下事情

給消息打上事務(wù)消息相關(guān)的標(biāo)記，用于MQ服務(wù)端區(qū)分普通消息和事務(wù)消息

發(fā)送半消息(half message)

發(fā)送成功則由transactionListener執(zhí)行本地事務(wù)

執(zhí)行endTransaction方法，如果半消息發(fā)送失敗或本地事務(wù)執(zhí)行失敗告訴服務(wù)端是刪除半消息，半消息發(fā)送成功且本地事務(wù)執(zhí)行成功則告訴服務(wù)端生效半消息。

發(fā)送半消息流程，Client端代碼到這里差不多就結(jié)束了，接下來看看RocketMQ Server端是如何處理的

RocketMQ Server

Server在接收到消息過后會進(jìn)行一些領(lǐng)域?qū)ο蟮霓D(zhuǎn)化和是否支持事務(wù)消息的權(quán)限校驗(yàn)，對理解事務(wù)消息用處不大，此處就省略對旁枝末節(jié)的介紹了。下面是TransactionalMessageBridge類處理half message的源碼

publicPutMessageResultputHalfMessage(MessageExtBrokerInner messageInner){returnstore.putMessage(parseHalfMessageInner(messageInner));}privateMessageExtBrokerInnerparseHalfMessageInner(MessageExtBrokerInner msgInner){? ? MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_TOPIC, msgInner.getTopic());? ? MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_QUEUE_ID,? ? ? ? String.valueOf(msgInner.getQueueId()));? ? msgInner.setSysFlag(? ? ? ? MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), MessageSysFlag.TRANSACTION_NOT_TYPE));? ? msgInner.setTopic(TransactionalMessageUtil.buildHalfTopic());? ? msgInner.setQueueId(0);? ? msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));returnmsgInner;}

這兩個(gè)方法主要做了以下事情：

publicclassMessageimplementsSerializable{privatestaticfinallongserialVersionUID =8445773977080406428L;privateString topic;privateintflag;privateMap properties;privatebyte[] body;privateString transactionId;}

將消息的topic，queueId放進(jìn)消息體自身的map里進(jìn)行緩存

將消息的topic 設(shè)置為“RMQ_SYS_TRANS_OP_HALF_TOPIC”，queueId設(shè)置為0

將消息寫入磁盤持久化

可以看到所有的事務(wù)半消息都會被放進(jìn)同一個(gè)topic的同一個(gè)queue里面，通過對topic的區(qū)分，從而避免了半消息被consumer給消費(fèi)到

Server將半消息持久化后然后會發(fā)送結(jié)果給我們本地的應(yīng)用程序。到了這里Server端對半消息的處理就結(jié)束了，緊接著的是定時(shí)任務(wù)的登場。

定時(shí)任務(wù)回查流程

RocketMQ Server

定時(shí)任務(wù)是一個(gè)叫TransactionalMessageService類的線程，下面是該類的check方法

@Overridepublicvoidcheck(longtransactionTimeout,inttransactionCheckMax,? ? AbstractTransactionalMessageCheckListener listener){? ? ? ? ? ? ? ? ? ...if(!putBackHalfMsgQueue(msgExt, i)) {continue;? ? }? ? ? listener.resolveHalfMsg(msgExt);? } ...}

check方法非常長，省略的代碼大致都是對半消息進(jìn)行過濾(如超過72小時(shí)的事務(wù)消息，就被算作過期)，只保留符合條件的半消息對其進(jìn)行回查。

其中很有意思的是putBackHalfMsgQueue方法，因?yàn)槊看伟寻胂拇疟P拉到內(nèi)存里進(jìn)行處理都會對其屬性進(jìn)行改變(例如TRANSACTION_CHECK_TIMES，這是是否丟棄事務(wù)消息的關(guān)鍵信息)，所以在發(fā)送回查消息之前需要對半消息再次放進(jìn)磁盤。RocketMQ采取的方法是基于最新的物理偏移量重新寫入，而不是對原有的半消息進(jìn)行修改，其中的目的就是RocketMQ的存儲設(shè)計(jì)采用順序?qū)?，如果去修改消?，無法做到高性能。

下面是resolveHalfMsg方法，主要就是開啟一個(gè)線程然后發(fā)送check消息。

publicvoidresolveHalfMsg(finalMessageExt msgExt){? ? executorService.execute(newRunnable() {@Overridepublicvoidrun(){try{? ? ? ? ? ? ? ? sendCheckMessage(msgExt);? ? ? ? ? ? }catch(Exception e) {? ? ? ? ? ? ? ? LOGGER.error("Send check message error!", e);? ? ? ? ? ? }? ? ? ? }? ? });}

本地應(yīng)用(client)

下面是DefaultMQProducerImpl的checkTransactionState方法，是本地應(yīng)用對回查消息的處理邏輯

@OverridepublicvoidcheckTransactionState(finalString addr,finalMessageExt msg,finalCheckTransactionStateRequestHeader header){? ? Runnable request =newRunnable() {? ? ? ? ...@Overridepublicvoidrun(){? ? ? ? ? ? ...? ? TransactionListener transactionListener = getCheckListener();? ? ? ? ? ? ...? ? localTransactionState = transactionListener.checkLocalTransaction(message);? ? ? ? ? ? ? ...this.processTransactionState(? ? ? ? ? ? ? ? ? ? localTransactionState,? ? ? ? ? ? ? ? ? ? group,? ? ? ? ? ? ? ? ? ? exception);? ? ? ? ? ? ? ? }privatevoidprocessTransactionState(? ? ? ? ? ... DefaultMQProducerImpl.this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, thisHeader, remark,3000);? ? ? ? ? ...? ? ? ? }? ? };this.checkExecutor.submit(request);}

精簡代碼邏輯后可以清晰的看到

開啟一個(gè)線程來執(zhí)行回查的邏輯

執(zhí)行transactionListener的checkLocalTransaction方法來獲取本地事務(wù)執(zhí)行的結(jié)果

RocketMQ Server

RocketMQ 服務(wù)器在收到Client發(fā)過來的Commit消息后會

讀出半消息——>恢復(fù)topic等原消息體的信息——>和普通消息一樣再次寫入磁盤——>刪除之前的半消息

如果是Rollback消息則直接刪除之前的半消息

到此，整條RocketMQ 事務(wù)消息的調(diào)用鏈就結(jié)束了

思考

1. 分布式事務(wù)等于事務(wù)消息嗎？

兩者并沒有關(guān)系，事務(wù)消息僅僅保證本地事務(wù)和MQ消息發(fā)送形成整體的原子性，而投遞到MQ服務(wù)器后，消費(fèi)者是否能一定消費(fèi)成功是無法保證的。

2. 源碼設(shè)計(jì)上有什么亮點(diǎn)嗎？

通過對整條鏈路源碼的學(xué)習(xí)理解發(fā)現(xiàn)還是有不少亮點(diǎn)的

server端回查消息的發(fā)送，client端回查消息邏輯的處理，client端commit/rollback消息的提交都是用了異步進(jìn)行，可以說能異步的地方都用了異步，通過異步+重試的方式保證了在分布式環(huán)境中即使短暫的網(wǎng)絡(luò)狀況不良好，也不會影響整體邏輯。

引入TransactionListener，真正做到了開閉原則以及依賴倒置原則，面向接口編程。整體擴(kuò)展性做得非常好，使用者只需要編寫自己的Listener就可以做到事務(wù)消息的發(fā)送，非常方便

TransactionMQProducer通過繼承DefaultMQProducer極大地復(fù)用了關(guān)于發(fā)送消息相關(guān)的邏輯

3. 源碼設(shè)計(jì)上有什么不足嗎？

RocketMQ作為一款極其成功的消息中間件，要發(fā)現(xiàn)不足不是那么容易了，筆者談幾點(diǎn)看法

sendMessageIntransaction等事務(wù)相關(guān)的方法被劃分在了DefaultMQProducer里面，從內(nèi)聚的角度來說這是跟事務(wù)相關(guān)的發(fā)送消息方法應(yīng)該被劃分在TransactionMQProducer。

所有topic的半消息都會寫在topic為RMQ_SYS_TRANS_OP_HALF_TOPIC的半消息隊(duì)列里，并且每條半消息，在整個(gè)鏈路里會被寫多次，如果并發(fā)很大且大部分消息都是事務(wù)消息的話，可靠性會存在問題。