本文將重點分析RocketMQ Broker如何處理事務(wù)消息提交、回滾命令，根據(jù)前面的介紹，其入口EndTransactionProcessor#proce***equest：

OperationResult result = new OperationResult();
if (MessageSysFlag.TRANSACTION_COMMIT_TYPE == requestHeader.getCommitOrRollback()) { // @1
result = this.brokerController.getTransactionalMessageService().commitMessage(requestHeader); // @2
if (result.getResponseCode() == ResponseCode.SUCCESS) { // @3
RemotingCommand res = checkPrepareMessage(result.getPrepareMessage(), requestHeader); // @4
if (res.getCode() == ResponseCode.SUCCESS) {
MessageExtBrokerInner msgInner = endMessageTransaction(result.getPrepareMessage()); // @5
msgInner.setSysFlag(MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), requestHeader.getCommitOrRollback()));
msgInner.setQueueOffset(requestHeader.getTranStateTableOffset());
msgInner.setPreparedTransactionOffset(requestHeader.getCommitLogOffset());
msgInner.setStoreTimestamp(result.getPrepareMessage().getStoreTimestamp()); // @6
RemotingCommand sendResult = sendFinalMessage(msgInner); // @7
if (sendResult.getCode() == ResponseCode.SUCCESS) {
this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage()); // @8
}
return sendResult;
}
return res;br/>}
}
代碼@1：如果請求為提交事務(wù)，進(jìn)入事務(wù)消息提交處理流程。
代碼@2：提交消息，別被這名字誤導(dǎo)了，該方法主要是根據(jù)commitLogOffset從commitlog文件中查找消息返回OperationResult實例：

private MessageExt prepareMessage ：消息對象。
private int responseCode：查找結(jié)果。
private String responseRemark ：錯誤提示。
代碼@3：如果成功查找到消息，則繼續(xù)處理，否則返回給客戶端，消息未找到錯誤信息。

代碼@4：驗證消息必要字段。
驗證消息的生產(chǎn)組與請求信息中的生產(chǎn)者組是否一致。
驗證消息的隊列偏移量（queueOffset）與請求信息中的偏移量是否一致。
驗證消息的commitLogOffset與請求信息中的CommitLogOffset是否一致。

代碼@5:調(diào)用endMessageTransaction方法，該方法主要的目的就是恢復(fù)事務(wù)消息的真實的主題、隊列，并設(shè)置事務(wù)ID。

代碼@6：設(shè)置消息的相關(guān)屬性，這一步應(yīng)該直接在endMessageTransaction中實現(xiàn)就好，統(tǒng)一恢復(fù)原消息的數(shù)量，特別關(guān)注的是取消了事務(wù)相關(guān)的系統(tǒng)標(biāo)記。

代碼@7：發(fā)送最終消息，其實現(xiàn)原理非常簡單，調(diào)用MessageStore將消息存儲在commitlog文件中，此時的消息，會被轉(zhuǎn)發(fā)到原消息主題對應(yīng)的消費(fèi)隊列，被消費(fèi)者消費(fèi)。

代碼@8：刪除預(yù)處理消息(prepare)，其實是將消息存儲在主題為：RMQ_SYS_TRANS_OP_HALF_TOPIC的主題中，代表這些消息已經(jīng)被處理（提交或回滾）。

上述就是事務(wù)消息提交的流程，事務(wù)回滾類似，接下來大概分析一下事務(wù)消息回滾的流程。

EndTransactionProcessor#proce***equest
else if (MessageSysFlag.TRANSACTION_ROLLBACK_TYPE == requestHeader.getCommitOrRollback()) {
result = this.brokerController.getTransactionalMessageService().rollbackMessage(requestHeader); // @1
if (result.getResponseCode() == ResponseCode.SUCCESS) {
RemotingCommand res = checkPrepareMessage(result.getPrepareMessage(), requestHeader);
if (res.getCode() == ResponseCode.SUCCESS) {
this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage()); // @2
}
return res;br/>}
}
代碼@1：回滾消息，其實內(nèi)部就是根據(jù)commitlogOffset查找消息。
代碼@2：將消息存儲在RMQ_SYS_TRANS_OP_HALF_TOPIC中，代表該消息已被處理，與提交事務(wù)消息不同的是，提交事務(wù)消息會將消息恢復(fù)原主題與隊列，再次存儲在commitlog文件中。

事務(wù)消息在Broker服務(wù)端的提交回滾流程就介紹到這了。其核心實現(xiàn)就是根據(jù)commitlogOffset找到消息，如果是提交動作，就恢復(fù)原消息的主題與隊列，再次存入commitlog文件進(jìn)而轉(zhuǎn)到消息消費(fèi)隊列，供消費(fèi)者消費(fèi)，然后將原預(yù)處理消息存入一個新的主題RMQ_SYS_TRANS_OP_HALF_TOPIC，代表該消息已被處理；回滾消息與提交事務(wù)消息不同的是，提交事務(wù)消息會將消息恢復(fù)原主題與隊列，再次存儲在commitlog文件中。

轉(zhuǎn)載于:https://blog.51cto.com/14031893/2340243

總結(jié)

以上是生活随笔為你收集整理的RocketMQ源码分析之RocketMQ事务消息实现原下篇(事务提交或回滚)的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。