深入淺出RocketMQ的事務消息
事務消息發(fā)送流程

半消息實現了分布式環(huán)境下的數據一致性的處理,生產者發(fā)送事務消息的流程如上圖所示,通過對源碼的學習,我們可以弄清楚下面幾點,也是半消息機制的核心:
1.為什么prepare消息不會被Consumer消費?
2.事務消息是如何提交和回滾的?
3.定時回查本地事務狀態(tài)的實現細節(jié)。
發(fā)送事務消息源碼分析
發(fā)送事務消息方法TransactionMQProducer.sendMessageInTransaction:
- msg:消息
- tranExecuter:本地事務執(zhí)行器
- arg:本地事務執(zhí)行器參數
public TransactionSendResult sendMessageInTransaction(final Message msg,
final LocalTransactionExecuter localTransactionExecuter, final Object arg)
throws MQClientException {
TransactionListener transactionListener = getCheckListener();
if (null == localTransactionExecuter && null == transactionListener) {
throw new MQClientException("tranExecutor is null", null);
}
// 忽視消息延遲的屬性
if (msg.getDelayTimeLevel() != 0) {
MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_DELAY_TIME_LEVEL);
}
Validators.checkMessage(msg, this.defaultMQProducer);
// 發(fā)送半消息
SendResult sendResult = null;
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_PRODUCER_GROUP, this.defaultMQProducer.getProducerGroup());
try {
sendResult = this.send(msg);
} catch (Exception e) {
throw new MQClientException("send message Exception", e);
}
// 處理發(fā)送半消息的結果
LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
Throwable localException = null;
switch (sendResult.getSendStatus()) {
// 發(fā)送半消息成功,執(zhí)行本地事務邏輯
case SEND_OK: {
try {
if (sendResult.getTransactionId() != null) {
msg.putUserProperty("__transactionId__", sendResult.getTransactionId());
}
String transactionId = msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
if (null != transactionId && !"".equals(transactionId)) {
msg.setTransactionId(transactionId);
}
// 執(zhí)行本地事務邏輯
if (null != localTransactionExecuter) {
localTransactionState = localTransactionExecuter.executeLocalTransactionBranch(msg, arg);
} else if (transactionListener != null) {
log.debug("Used new transaction API");
localTransactionState = transactionListener.executeLocalTransaction(msg, arg);
}
if (null == localTransactionState) {
localTransactionState = LocalTransactionState.UNKNOW;
}
if (localTransactionState != LocalTransactionState.COMMIT_MESSAGE) {
log.info("executeLocalTransactionBranch return {}", localTransactionState);
log.info(msg.toString());
}
} catch (Throwable e) {
log.info("executeLocalTransactionBranch exception", e);
log.info(msg.toString());
localException = e;
}
}
break;
// 發(fā)送半消息失敗,標記本地事務狀態(tài)為回滾
case FLUSH_DISK_TIMEOUT:
case FLUSH_SLAVE_TIMEOUT:
case SLAVE_NOT_AVAILABLE:
localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
break;
default:
break;
}
// 結束事務,設置消息 COMMIT / ROLLBACK
try {
this.endTransaction(msg, sendResult, localTransactionState, localException);
} catch (Exception e) {
log.warn("local transaction execute " + localTransactionState + ", but end broker transaction failed", e);
}
// 返回事務發(fā)送結果
TransactionSendResult transactionSendResult = new TransactionSendResult();
transactionSendResult.setSendStatus(sendResult.getSendStatus());
transactionSendResult.setMessageQueue(sendResult.getMessageQueue());
// 提取Prepared消息的uniqID
transactionSendResult.setMsgId(sendResult.getMsgId());
transactionSendResult.setQueueOffset(sendResult.getQueueOffset());
transactionSendResult.setTransactionId(sendResult.getTransactionId());
transactionSendResult.setLocalTransactionState(localTransactionState);
return transactionSendResult;
}該方法的入參包含有一個需要用戶實現本地事務的LocalTransactionExecuter executer,executer中會進行事務操作以保證本地事務和消息發(fā)送這兩個操作的原子性。
由上面的源碼可知:
Producer會首先發(fā)送一個半消息到Broker中:
- 半消息發(fā)送成功,執(zhí)行事務
- 半消息發(fā)送失敗,不執(zhí)行事務
半消息發(fā)送到Broker后不會被Consumer消費掉的原因有以下兩點:
- Broker在將消息寫入CommitLog時會判斷消息類型,如果是prepare或者rollback消息,ConsumeQueue的offset不變
- Broker在構造ConsumeQueue時會判斷是否是處于prepare或者rollback狀態(tài)的消息,如果是則不會將該消息放入ConsumeQueue里,Consumer在拉取消息時也就不會拉取到這條消息
Producer會根據半消息的發(fā)送結果和本地任務執(zhí)行結果來決定如何處理事務(commit或rollback),方法最后調用了endTransaction來處理事務的執(zhí)行結果,源碼如下:
- sendResult:發(fā)送半消息的結果
- localTransactionState:本地事務狀態(tài)
- localException:執(zhí)行本地事務邏輯產生的異常
- RemotingException:遠程調用異常
- MQBrokerException:Broker異常
- InterruptedException:當線程中斷異常
- UnknownHostException:未知host異常
public void endTransaction(
final Message msg,
final SendResult sendResult,
final LocalTransactionState localTransactionState,
final Throwable localException) throws RemotingException, MQBrokerException, InterruptedException, UnknownHostException {
// 解碼消息id
final MessageId id;
if (sendResult.getOffsetMsgId() != null) {
id = MessageDecoder.decodeMessageId(sendResult.getOffsetMsgId());
} else {
id = MessageDecoder.decodeMessageId(sendResult.getMsgId());
}
// 創(chuàng)建請求
String transactionId = sendResult.getTransactionId();
final String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(sendResult.getMessageQueue().getBrokerName());
EndTransactionRequestHeader requestHeader = new EndTransactionRequestHeader();
requestHeader.setTransactionId(transactionId);
requestHeader.setCommitLogOffset(id.getOffset());
switch (localTransactionState) {
case COMMIT_MESSAGE:
requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_COMMIT_TYPE);
break;
case ROLLBACK_MESSAGE:
requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_ROLLBACK_TYPE);
break;
case UNKNOW:
requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_NOT_TYPE);
break;
default:
break;
}
doExecuteEndTransactionHook(msg, sendResult.getMsgId(), brokerAddr, localTransactionState, false);
requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
requestHeader.setTranStateTableOffset(sendResult.getQueueOffset());
requestHeader.setMsgId(sendResult.getMsgId());
String remark = localException != null ? ("executeLocalTransactionBranch exception: " + localException.toString()) : null;
// 提交 commit / rollback 消息
this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, requestHeader, remark,
this.defaultMQProducer.getSendMsgTimeout());
}該方法是將事務執(zhí)行的結果發(fā)送給Broker,再由Broker決定是否進行消息投遞,執(zhí)行步驟如下:
1.收到消息后先檢查是否是事務消息,如果不是事務消息則直接返回
2.根據請求頭里的offset查詢半消息,如果查詢結果為空則直接返回
3.根據半消息構造新消息,新構造的消息會被重新寫入到CommitLog里,rollback消息的消息體為空
4.如果是rollback消息,則該消息不會被投遞
具體原因上文中已經分析過:只有commit消息才會被Broker投遞給consumer
RocketMQ會將commit消息和rollback消息都寫入到commitLog里,但rollback消息的消息體為空且不會被投遞,CommitLog在刪除過期消息時才會將其刪除。當事務commit成功之后,RocketMQ會重新封裝半消息并將其投遞給Consumer端消費。
事務消息回查
Broker發(fā)起
相較于普通消息,事務消息主要依賴下面三個類:
1.TransactionStateService:事務狀態(tài)服務,負責對事務消息進行管理,包括存儲和更新事務消息狀態(tài)、回查狀態(tài)等
2.TranStateTable:事務消息狀態(tài)存儲表,基于MappedFileQueue實現
3.TranRedoLog:TranStateTable的日志,每次寫入操作都會記錄日志,當Broker宕機時,可以利用這個文件做數據恢復
存儲半消息到CommitLog時,使用offset索引到對應的TranStateTable的位置
到此這篇關于深入淺出RocketMQ的事務消息的文章就介紹到這了,更多相關RocketMQ事務消息內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家!
相關文章
SpringBoot使用SchedulingConfigurer實現多個定時任務多機器部署問題(推薦)
這篇文章主要介紹了SpringBoot使用SchedulingConfigurer實現多個定時任務多機器部署問題,定時任務多機器部署解決方案,方式一拆分,單獨拆分出來,單獨跑一個應用,方式二是基于aop攔截處理(搶占執(zhí)行),只要有一個執(zhí)行,其它都不執(zhí)行,需要的朋友可以參考下2023-01-01
Mybatis in條件傳參的三種實現方式(直接$,List,[])
這篇文章主要介紹了Mybatis in條件傳參的三種實現方式(直接$,List,[]),具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-12-12

