RocketMQ事務(wù)消息接口介紹
當我們在業(yè)務(wù)邏輯中發(fā)送消息時，消息與業(yè)務(wù)的事務(wù)之間難以保證一致性，如果業(yè)務(wù)代碼出現(xiàn)異常，如果已發(fā)送的消息無法回滾，則很會出現(xiàn)數(shù)據(jù)不一致的情況，RocketMQ的事務(wù)消息支持在業(yè)務(wù)邏輯與發(fā)送消息之間提供事務(wù)保證，RocketMQ通過兩階段的方式提供事務(wù)消息的支持。
RocketMQ實現(xiàn)事務(wù)消息依賴于TransactionListener接口，此接口的定義如下：
其中包含兩個方法：
- executeLocalTransaction方法會在發(fā)送消息后調(diào)用，用于執(zhí)行本地事務(wù)，如果本地事務(wù)執(zhí)行成功，rocketmq再提交消息
- checkLocalTransaction用于對本地事務(wù)做檢查，rocketmq依賴此方法做補覺，后文再細說
以官方的示例為例子，我們看看如何使用RocketMQ的事務(wù)消息，首先實現(xiàn)一個TransactionListener:
然后我們再通過實現(xiàn)的TransactionListenerImpl類創(chuàng)建TransactionMQProducer，所有事務(wù)消息都需要通過TransactionMQProducer發(fā)送：
最后發(fā)送消息：
消息的消費和普通消息一樣，這里不多說了。下面我們再看細說一下RocketMQ的事務(wù)消息的實現(xiàn)機制
事務(wù)消息的執(zhí)行機制
如上圖所示，RocketMQ通過兩個內(nèi)部的topic來實現(xiàn)對消息的兩階段支持，RocketMQ在實現(xiàn)事務(wù)消息時，實際上是通過將生產(chǎn)投遞過來的消息（消息上帶有事務(wù)標識）投遞到一個名為RMS_SYS_TRANS_HALF_TOPIC的topic中，而不是投遞到真正的topic中，這個過程是第一階段(prepare)，然后producer再通過TransactionListener的executeLocalTransaction方法執(zhí)行本地事務(wù)，當producer的localTransaction處理成功或者失敗后，producer會向broker發(fā)送commit或rollback命令，如果是commit，則broker會將投遞到RMQ_SYS_TRANS_HALF_TOPIC中的消息投遞到真實的topic中，然后再投遞一個表示刪除的消息到RMQ_SYS_TRANS_OP_HALF_TOPIC中，表示當前事務(wù)已完成；如果是rollback，則沒有投遞到真實topic的過程，只需要投遞表示刪除的消息到RMQ_SYS_TRANS_OP_HALF_TOPIC。最后，消費者和消費普通的消息一樣消費事務(wù)消息。
整個過程如果沒有遇到問題，則一切OK，但整個過程中可能會遇到以下錯誤：
- 第一階段（prepare）失?。航o應用返回發(fā)送消息失敗
- 事務(wù)失?。喊l(fā)送回滾命令給broker，由broker執(zhí)行消息的回滾
- Commit或rollback失?。河蒪roker定時向producer發(fā)起事務(wù)檢查，如果本地事務(wù)成功，則提交消息事務(wù)，否則回滾消息事務(wù)
事務(wù)狀態(tài)的檢查有兩種情況：
- commit/rollback：broker會執(zhí)行相應的commit/rollback操作
- 如果是TRANSACTION_NOT_TYPE，則一段時間后會再次檢查，當檢查的次數(shù)超過上限（默認15次）則丟棄消息
異常情況示意圖如下：
源碼閱讀
最后我們看看幾處關(guān)鍵代碼，首先是producer的發(fā)送消息部分，在DefaultMQMessageImpl類的sendMessageInTransaction方法中使用了丙階段的方式處理事務(wù)：
發(fā)送prepare消息成功后表示第一階段成功，然后再調(diào)用transactionListener.executeLocalTransaction執(zhí)行本地事務(wù)，隨便根據(jù)本地事務(wù)的執(zhí)行結(jié)果調(diào)用endTransaction方法做第二階段的處理：
接下來看看broker是如何處理兩階段的，首先我們看看prepare的處理，在SendMessageProcessor類的sendMessage方法中，我們可以看到獲取事務(wù)標識并決定處理邏輯的代碼：
如果事務(wù)標識為true，則調(diào)用TransactionalMessageService的prepareMessage方法，我們可以進入到此方法中，一直到TransactionalMessageBridge的parseHalfMessage方法，并最終找到消息的處理方式：
parseHalfMessage方法先將消息真實的topic和queueId加到到property里，然后將消息的topic設(shè)置成TransactionalMessageUtil.buildHalfTopic()調(diào)用的返回值，返回的topic正是RMQ_SYS_TRANS_HALF_TOPIC，這里對topic作了一個轉(zhuǎn)換，因此在一階段完成后，消費者還無法消費到事務(wù)消息
我們再來看看二階段的處理方式，進入到EndTransactionProcessor類的processRequest方法中，可以看到如下代碼：
其中兩個核心代碼的調(diào)用，一個是commit時調(diào)用的sendFinalMessage方法用于將消息投遞到真實的topic中，另一個是TransactionalMessageService的deletePrepareMessage方法用于投遞一個用于標識當前事務(wù)的一階段消息為刪除的消息，在看sendFinalMessage方法的實現(xiàn)前，我們先看一下在此方法調(diào)用前調(diào)用的endTransactionMessage方法的實現(xiàn)：
可以看到關(guān)鍵代碼，將消息的topic和queueId設(shè)置回真實的topic和queueId，然后在sendFinalMessage中存儲消息：
我們再來看看TransactionalMessageService的deletePrepareMessage方法的實現(xiàn)，很明顯，是新寫入了一個消息：
這里可以看到，新寫入的消息的tag是REMOVETAG，我們進入到putOpMessage方法，在addRemoveTagInTransactionOp方法的調(diào)用中可以看到使用的topic是TransactionalMessageUtil.buildOpTopic()的返回值 ，即RMQ_SYS_TRANS_OP_HALF_TOPIC
最后，我們再來看看本地事務(wù)的check相關(guān)的代碼，我們進入到TransactionalMessageCheckService類中，此類包含一個線程，此線程默認每分鐘觸發(fā)一次事務(wù)檢查，在其onWaitEnd方法中，可以看到實際上還是調(diào)用了TransactionalMessageService的check方法：
這里默認的timeout是6秒和checkMax是15，表示的意思是6秒以上沒commit/rollback的消息才做事務(wù)檢查，檢查次數(shù)越過15次則丟棄事務(wù)，我們可以進入到TransactionalMessageService的check方法中，其中有一大段的邏輯用于判斷一個消息是否應該做事務(wù)檢查，這里不解釋了，我們直接看觸發(fā)事務(wù)檢查的代碼：
很明顯，listener.resolveHalfMsg方法用于觸發(fā)事務(wù)的檢查，其實現(xiàn)如下：
可以看到，它使用了broker到client的調(diào)用，觸發(fā)producer的事務(wù)檢查，至于事務(wù)檢查如何處理，我們可以回到producer的DefaultMQProducerImpl類中，其中的checkTransactionState方法調(diào)用了TransactionListener的checkLocalTransaction方法用于處理事務(wù)的檢查：
最后事務(wù)檢查的結(jié)果會由processTransactionState方法做處理：這里和前文講過的事務(wù)消息的第二階段處理的代碼一樣，將事務(wù)結(jié)果發(fā)送到broker，并由broker的EndTransactionProcessor的processRequest做事務(wù)二階段的處理
最后說一句，雖然RocketMQ的代碼寫的不優(yōu)雅，而且for/if-else等嵌套非常深，但是理解了它的運行機制后還是能有所收獲的。
