RocketMQ事務(wù)消息接口介紹
當(dāng)我們?cè)跇I(yè)務(wù)邏輯中發(fā)送消息時(shí)，消息與業(yè)務(wù)的事務(wù)之間難以保證一致性张足，如果業(yè)務(wù)代碼出現(xiàn)異常物臂，如果已發(fā)送的消息無(wú)法回滾颇蜡，則很會(huì)出現(xiàn)數(shù)據(jù)不一致的情況，RocketMQ的事務(wù)消息支持在業(yè)務(wù)邏輯與發(fā)送消息之間提供事務(wù)保證假颇，RocketMQ通過(guò)兩階段的方式提供事務(wù)消息的支持胚鸯。
RocketMQ實(shí)現(xiàn)事務(wù)消息依賴于TransactionListener接口，此接口的定義如下：
其中包含兩個(gè)方法：
- executeLocalTransaction方法會(huì)在發(fā)送消息后調(diào)用笨鸡，用于執(zhí)行本地事務(wù)姜钳，如果本地事務(wù)執(zhí)行成功，rocketmq再提交消息
- checkLocalTransaction用于對(duì)本地事務(wù)做檢查形耗，rocketmq依賴此方法做補(bǔ)覺(jué)哥桥，后文再細(xì)說(shuō)
以官方的示例為例子，我們看看如何使用RocketMQ的事務(wù)消息激涤，首先實(shí)現(xiàn)一個(gè)TransactionListener:
然后我們?cè)偻ㄟ^(guò)實(shí)現(xiàn)的TransactionListenerImpl類創(chuàng)建TransactionMQProducer拟糕，所有事務(wù)消息都需要通過(guò)TransactionMQProducer發(fā)送：
最后發(fā)送消息：
消息的消費(fèi)和普通消息一樣，這里不多說(shuō)了倦踢。下面我們?cè)倏醇?xì)說(shuō)一下RocketMQ的事務(wù)消息的實(shí)現(xiàn)機(jī)制
事務(wù)消息的執(zhí)行機(jī)制
如上圖所示送滞，RocketMQ通過(guò)兩個(gè)內(nèi)部的topic來(lái)實(shí)現(xiàn)對(duì)消息的兩階段支持，RocketMQ在實(shí)現(xiàn)事務(wù)消息時(shí)辱挥，實(shí)際上是通過(guò)將生產(chǎn)投遞過(guò)來(lái)的消息（消息上帶有事務(wù)標(biāo)識(shí)）投遞到一個(gè)名為RMS_SYS_TRANS_HALF_TOPIC的topic中犁嗅，而不是投遞到真正的topic中，這個(gè)過(guò)程是第一階段(prepare)晤碘，然后producer再通過(guò)TransactionListener的executeLocalTransaction方法執(zhí)行本地事務(wù)褂微，當(dāng)producer的localTransaction處理成功或者失敗后，producer會(huì)向broker發(fā)送commit或rollback命令园爷，如果是commit宠蚂，則broker會(huì)將投遞到RMQ_SYS_TRANS_HALF_TOPIC中的消息投遞到真實(shí)的topic中，然后再投遞一個(gè)表示刪除的消息到RMQ_SYS_TRANS_OP_HALF_TOPIC中腮介，表示當(dāng)前事務(wù)已完成肥矢；如果是rollback，則沒(méi)有投遞到真實(shí)topic的過(guò)程叠洗，只需要投遞表示刪除的消息到RMQ_SYS_TRANS_OP_HALF_TOPIC甘改。最后，消費(fèi)者和消費(fèi)普通的消息一樣消費(fèi)事務(wù)消息灭抑。
整個(gè)過(guò)程如果沒(méi)有遇到問(wèn)題十艾，則一切OK，但整個(gè)過(guò)程中可能會(huì)遇到以下錯(cuò)誤：
- 第一階段（prepare）失斕诮凇：給應(yīng)用返回發(fā)送消息失敗
- 事務(wù)失斖怠：發(fā)送回滾命令給broker荤牍，由broker執(zhí)行消息的回滾
- Commit或rollback失敗：由broker定時(shí)向producer發(fā)起事務(wù)檢查庆冕，如果本地事務(wù)成功康吵，則提交消息事務(wù)，否則回滾消息事務(wù)
事務(wù)狀態(tài)的檢查有兩種情況：
- commit/rollback：broker會(huì)執(zhí)行相應(yīng)的commit/rollback操作
- 如果是TRANSACTION_NOT_TYPE访递，則一段時(shí)間后會(huì)再次檢查晦嵌，當(dāng)檢查的次數(shù)超過(guò)上限（默認(rèn)15次）則丟棄消息
異常情況示意圖如下：
源碼閱讀
最后我們看看幾處關(guān)鍵代碼，首先是producer的發(fā)送消息部分拷姿，在DefaultMQMessageImpl類的sendMessageInTransaction方法中使用了丙階段的方式處理事務(wù)：
發(fā)送prepare消息成功后表示第一階段成功惭载，然后再調(diào)用transactionListener.executeLocalTransaction執(zhí)行本地事務(wù)，隨便根據(jù)本地事務(wù)的執(zhí)行結(jié)果調(diào)用endTransaction方法做第二階段的處理：
接下來(lái)看看broker是如何處理兩階段的响巢，首先我們看看prepare的處理描滔，在SendMessageProcessor類的sendMessage方法中，我們可以看到獲取事務(wù)標(biāo)識(shí)并決定處理邏輯的代碼：
如果事務(wù)標(biāo)識(shí)為true踪古，則調(diào)用TransactionalMessageService的prepareMessage方法含长，我們可以進(jìn)入到此方法中，一直到TransactionalMessageBridge的parseHalfMessage方法灾炭，并最終找到消息的處理方式：
parseHalfMessage方法先將消息真實(shí)的topic和queueId加到到property里茎芋，然后將消息的topic設(shè)置成TransactionalMessageUtil.buildHalfTopic()調(diào)用的返回值，返回的topic正是RMQ_SYS_TRANS_HALF_TOPIC蜈出，這里對(duì)topic作了一個(gè)轉(zhuǎn)換田弥，因此在一階段完成后，消費(fèi)者還無(wú)法消費(fèi)到事務(wù)消息
我們?cè)賮?lái)看看二階段的處理方式铡原，進(jìn)入到EndTransactionProcessor類的processRequest方法中偷厦，可以看到如下代碼：
其中兩個(gè)核心代碼的調(diào)用，一個(gè)是commit時(shí)調(diào)用的sendFinalMessage方法用于將消息投遞到真實(shí)的topic中燕刻，另一個(gè)是TransactionalMessageService的deletePrepareMessage方法用于投遞一個(gè)用于標(biāo)識(shí)當(dāng)前事務(wù)的一階段消息為刪除的消息只泼，在看sendFinalMessage方法的實(shí)現(xiàn)前，我們先看一下在此方法調(diào)用前調(diào)用的endTransactionMessage方法的實(shí)現(xiàn)：
可以看到關(guān)鍵代碼卵洗，將消息的topic和queueId設(shè)置回真實(shí)的topic和queueId请唱，然后在sendFinalMessage中存儲(chǔ)消息：
我們?cè)賮?lái)看看TransactionalMessageService的deletePrepareMessage方法的實(shí)現(xiàn)，很明顯过蹂，是新寫(xiě)入了一個(gè)消息：
這里可以看到十绑，新寫(xiě)入的消息的tag是REMOVETAG，我們進(jìn)入到putOpMessage方法酷勺，在addRemoveTagInTransactionOp方法的調(diào)用中可以看到使用的topic是TransactionalMessageUtil.buildOpTopic()的返回值 本橙，即RMQ_SYS_TRANS_OP_HALF_TOPIC
最后，我們?cè)賮?lái)看看本地事務(wù)的check相關(guān)的代碼脆诉，我們進(jìn)入到TransactionalMessageCheckService類中甚亭，此類包含一個(gè)線程贷币，此線程默認(rèn)每分鐘觸發(fā)一次事務(wù)檢查，在其onWaitEnd方法中亏狰，可以看到實(shí)際上還是調(diào)用了TransactionalMessageService的check方法：
這里默認(rèn)的timeout是6秒和checkMax是15役纹，表示的意思是6秒以上沒(méi)commit/rollback的消息才做事務(wù)檢查，檢查次數(shù)越過(guò)15次則丟棄事務(wù)暇唾，我們可以進(jìn)入到TransactionalMessageService的check方法中字管，其中有一大段的邏輯用于判斷一個(gè)消息是否應(yīng)該做事務(wù)檢查，這里不解釋了信不，我們直接看觸發(fā)事務(wù)檢查的代碼：
很明顯，listener.resolveHalfMsg方法用于觸發(fā)事務(wù)的檢查亡呵，其實(shí)現(xiàn)如下：
可以看到抽活，它使用了broker到client的調(diào)用，觸發(fā)producer的事務(wù)檢查锰什，至于事務(wù)檢查如何處理下硕，我們可以回到producer的DefaultMQProducerImpl類中，其中的checkTransactionState方法調(diào)用了TransactionListener的checkLocalTransaction方法用于處理事務(wù)的檢查：
最后事務(wù)檢查的結(jié)果會(huì)由processTransactionState方法做處理：這里和前文講過(guò)的事務(wù)消息的第二階段處理的代碼一樣汁胆，將事務(wù)結(jié)果發(fā)送到broker梭姓，并由broker的EndTransactionProcessor的processRequest做事務(wù)二階段的處理
最后說(shuō)一句，雖然RocketMQ的代碼寫(xiě)的不優(yōu)雅嫩码，而且for/if-else等嵌套非常深誉尖，但是理解了它的運(yùn)行機(jī)制后還是能有所收獲的。
