分布式場(chǎng)景下的事務(wù)機(jī)制
事務(wù)消息是RocketMQ的一個(gè)非常特色的高級(jí)特性，它的基礎(chǔ)訴求是通過(guò)RocketMQ的事務(wù)機(jī)制，來(lái)保證上下游的數(shù)據(jù)?致性。
我們?cè)趩螜C(jī)版本下面只需要在業(yè)務(wù)方法上加上對(duì)應(yīng)的事務(wù)就可以達(dá)到效果，但是分布式的場(chǎng)景下，多個(gè)系統(tǒng)之間的協(xié)調(diào)配合，你無(wú)法知道到底是那個(gè)先執(zhí)行那個(gè)后執(zhí)行，當(dāng)然在微服務(wù)里面存在Seate框架來(lái)保證事務(wù)，但是這事務(wù)的保證始終是心頭大患，只能用一句話(huà)形容魚(yú)和熊掌不可兼得。
而RocketMq的事務(wù)消息能夠在提升性能的情況下滿(mǎn)足要求，其主要實(shí)現(xiàn)是支持分布式情況下保障消息生產(chǎn)和本地事務(wù)的最終一致性，消息生產(chǎn)我們可以使用順序消息去執(zhí)行，這樣我們只需要滿(mǎn)足這兩個(gè)的事務(wù)即可。
實(shí)現(xiàn)過(guò)程
圖片
準(zhǔn)備階段：生產(chǎn)者將消息發(fā)送到Broker，Broker向生產(chǎn)者發(fā)送ack表示消息發(fā)送成功，但是此時(shí)的消息為一個(gè)等待狀態(tài)，不會(huì)被消費(fèi)者去消費(fèi)。（生產(chǎn)者繼續(xù)執(zhí)行接下來(lái)的代碼）
確認(rèn)階段：當(dāng)我們執(zhí)行完所有的代碼后，本地事務(wù)要么回滾要么提交，此時(shí)當(dāng)我們了解本地事務(wù)的狀態(tài)后，將結(jié)果推送給Broker做二次確認(rèn)結(jié)果，如果為Commit則將修改激活準(zhǔn)備推送給消費(fèi)者，如果為Rollback則將消息進(jìn)行回滾。
補(bǔ)償機(jī)制：當(dāng)出現(xiàn)異常情況沒(méi)有發(fā)生二次確認(rèn)，此時(shí)我們?cè)诠潭〞r(shí)間后將會(huì)進(jìn)行回查，檢查回查消息對(duì)應(yīng)的本地事務(wù)的狀態(tài)，重寫(xiě)Commit或者Rollback。
涉及狀態(tài)以及注意點(diǎn)
事務(wù)消息存在三種狀態(tài)：
CommitTransaction：提交事務(wù)狀態(tài)，此狀態(tài)下允許消費(fèi)者消費(fèi)。
RollbackTransaction：回滾事務(wù)狀態(tài)，此狀態(tài)下消息會(huì)被刪除。
Unknown：中間狀態(tài)，此狀態(tài)下會(huì)等待本地事務(wù)處理結(jié)果進(jìn)行對(duì)應(yīng)操作。
注意點(diǎn)：
本消息狀態(tài)是一種對(duì)消費(fèi)者不可見(jiàn)的狀態(tài)，將消息的內(nèi)容放到系統(tǒng)Topic的RMQ_SYS_TRANS_HALF_TOPIC隊(duì)列里面去。
事務(wù)消息中的相關(guān)參數(shù)可以進(jìn)行設(shè)置，比如：本地事務(wù)回查次數(shù)transactionCheckMax默認(rèn)15次，本地事務(wù)回查的間隙transactionCheckInterval默認(rèn)60s，超出后會(huì)直接將消息丟棄。
RocketMQ的事務(wù)消息是指應(yīng)用本地事務(wù)和發(fā)送消息操作可以定義到全局事務(wù)中，要么同時(shí)成功，要么同時(shí)失敗，通過(guò)RocketMQ的事務(wù)信息可以實(shí)現(xiàn)可靠消息的最終一致性方案。
源碼解析
Producer端通過(guò)構(gòu)建TransactionMQProducer對(duì)象綁定事務(wù)監(jiān)聽(tīng)。
TransactionListener transactionListener = new TransactionListener() { @Override public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { return LocalTransactionState.COMMIT_MESSAGE; }
@Override public LocalTransactionState checkLocalTransaction(MessageExt msg) { return LocalTransactionState.COMMIT_MESSAGE; }};TransactionMQProducer producer = new TransactionMQProducer(producerGroupTemp);producer.setTransactionListener(transactionListener);producer.setNamesrvAddr("127.0.0.1:9876");product.start();SendResult result = producer.sendMessageInTransaction(message, arg);執(zhí)行sendMessageInTransaction方法來(lái)發(fā)送消息。
public TransactionSendResult sendMessageInTransaction(final Message msg,
final LocalTransactionExecuter localTransactionExecuter, final Object arg)
throws MQClientException {
// 檢查T(mén)ransactionListener是否存在，如果不存在就直接拋異常
TransactionListener transactionListener = getCheckListener();
if (null == localTransactionExecuter && null == transactionListener) {
throw new MQClientException("tranExecutor is null", null);
}
// 事務(wù)消息不支持延遲等特性
if (msg.getDelayTimeLevel() != 0) {
MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_DELAY_TIME_LEVEL);
}
Validators.checkMessage(msg, this.defaultMQProducer);
SendResult sendResult = null;
// 設(shè)置half屬性，表明是事務(wù)屬性
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
// 設(shè)置所屬生成者組
// broker向生產(chǎn)者發(fā)送回查事務(wù)請(qǐng)求根據(jù)這個(gè)producergroup找到指定的channel
// 生產(chǎn)者能找到所有在同一個(gè)組的機(jī)器實(shí)例從而檢查事務(wù)狀態(tài)
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_PRODUCER_GROUP, this.defaultMQProducer.getProducerGroup());
try {
// 同步發(fā)送
sendResult = this.send(msg);
} catch (Exception e) {
throw new MQClientException("send message Exception", e);
}
LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
Throwable localException = null;
// 消息返回信息
switch (sendResult.getSendStatus()) {
// 第一階段消息發(fā)送成功
case SEND_OK: {
try {
if (sendResult.getTransactionId() != null) {
// 設(shè)置事務(wù)ID屬性
msg.putUserProperty("__transactionId__", sendResult.getTransactionId());
}
String transactionId = msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
if (null != transactionId && !"".equals(transactionId)) {
msg.setTransactionId(transactionId);
}
if (null != localTransactionExecuter) {
// 執(zhí)行本地事務(wù)
localTransactionState = localTransactionExecuter.executeLocalTransactionBranch(msg, arg);
} else if (transactionListener != null) {
log.debug("Used new transaction API");
// 發(fā)送消息成功后，執(zhí)行本地操作
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;
case FLUSH_DISK_TIMEOUT:
case FLUSH_SLAVE_TIMEOUT:
case SLAVE_NOT_AVAILABLE:
localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
break;
default:
break;
}
try {
// 本地事務(wù)執(zhí)行完畢向broker提交事務(wù)或回滾事務(wù)
this.endTransaction(msg, sendResult, localTransactionState, localException);
} catch (Exception e) {
log.warn("local transaction execute " + localTransactionState + ", but end broker transaction failed", e);
}
TransactionSendResult transactionSendResult = new TransactionSendResult();
transactionSendResult.setSendStatus(sendResult.getSendStatus());
transactionSendResult.setMessageQueue(sendResult.getMessageQueue());
transactionSendResult.setMsgId(sendResult.getMsgId());
transactionSendResult.setQueueOffset(sendResult.getQueueOffset());
transactionSendResult.setTransactionId(sendResult.getTransactionId());
transactionSendResult.setLocalTransactionState(localTransactionState);
return transactionSendResult;
}首先發(fā)送第一階段信息直接返回半提交狀態(tài)，然后執(zhí)行本地事務(wù)返回事務(wù)的三種狀態(tài)，未知，回滾，提交，最后執(zhí)行endTransaction方法，把事務(wù)執(zhí)行的狀態(tài)告訴broker。
endTransaction方法
根據(jù)本地事務(wù)執(zhí)行狀態(tài)構(gòu)建requestHeader對(duì)象執(zhí)行二階段提交。
public void endTransaction(
final Message msg,
final SendResult sendResult,
final LocalTransactionState localTransactionState,
final Throwable localException) throws RemotingException, MQBrokerException, InterruptedException, UnknownHostException {
final MessageId id;
// 獲取消息中的MessageId
if (sendResult.getOffsetMsgId() != null) {
id = MessageDecoder.decodeMessageId(sendResult.getOffsetMsgId());
} else {
id = MessageDecoder.decodeMessageId(sendResult.getMsgId());
}
String transactionId = sendResult.getTransactionId();
// 找到broker地址
final String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(sendResult.getMessageQueue().getBrokerName());
// 構(gòu)建EndTransactionRequestHeader對(duì)象
EndTransactionRequestHeader requestHeader = new EndTransactionRequestHeader();
requestHeader.setTransactionId(transactionId);
// offset是prepare消息中offsetMsgId中獲取的
requestHeader.setCommitLogOffset(id.getOffset());
requestHeader.setBname(sendResult.getMessageQueue().getBrokerName());
// 社會(huì)提交/回滾狀態(tài)
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;
// 發(fā)送給broker端
this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, requestHeader, remark,
this.defaultMQProducer.getSendMsgTimeout());
}將本地方法執(zhí)行事務(wù)的結(jié)果發(fā)送給Broker，通過(guò)endTransactionOneway方法創(chuàng)建Code為END_TRANSACTION的消息，然后在Broker就會(huì)找出對(duì)應(yīng)的Processor來(lái)處理。
Broker端處理
Broker總共存在兩個(gè)處理，首先針對(duì)第一個(gè)階段發(fā)送的Half消息，broker要進(jìn)行相關(guān)的操作，后面endTransaction提交進(jìn)來(lái)的事務(wù)狀態(tài)，針對(duì)三種狀態(tài)進(jìn)行相關(guān)操作。
接收第一階段發(fā)送的Half消息
SendMessageProcessor的sendMessage方法中去執(zhí)行處理事務(wù)消息。
// 發(fā)送Half消息時(shí)，在屬性中設(shè)置了PROPERTY_TRANSACTION_PREPARED為true，這里根據(jù)這個(gè)屬性判斷是否是事務(wù)消息
String traFlag = oriProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);
if (Boolean.parseBoolean(traFlag)
&& !(msgInner.getReconsumeTimes() > 0 && msgInner.getDelayTimeLevel() > 0)) { //For client under version 4.6.1
if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) {
response.setCode(ResponseCode.NO_PERMISSION);
response.setRemark(
"the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1()
+ "] sending transaction message is forbidden");
return response;
}
// 事務(wù)消息進(jìn)入這里，把消息的topic改成RMQ_SYS_TRANS_HALF_TOPIC，以同步刷盤(pán)的方式存入store
putMessageResult = this.brokerController.getTransactionalMessageService().prepareMessage(msgInner);
}如果消息攜帶事務(wù)標(biāo)記就去執(zhí)行TransactionMessageService類(lèi)的prepareMessage方法進(jìn)行相關(guān)的處理。
// 解析Half消息
private MessageExtBrokerInner parseHalfMessageInner(MessageExtBrokerInner msgInner) {
// 把真實(shí)的topic和真實(shí)的queueId放在消息的屬性中
MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_TOPIC, msgInner.getTopic());
MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_QUEUE_ID,
String.valueOf(msgInner.getQueueId()));
// 設(shè)置默認(rèn)的事務(wù)狀態(tài)為T(mén)RANSACTION_NOT_TYPE=>unknow
msgInner.setSysFlag(
MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), MessageSysFlag.TRANSACTION_NOT_TYPE));
// 將消息的topic設(shè)置為RMQ_SYS_TRANS_HALF_TOPIC，這個(gè)是對(duì)消費(fèi)者不可見(jiàn)的
msgInner.setTopic(TransactionalMessageUtil.buildHalfTopic());
// 設(shè)置queueId=0
msgInner.setQueueId(0);
msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
return msgInner;
}進(jìn)行topic的切換，將原來(lái)的topic存入到消息的屬性里面，將消息的topic設(shè)置為RMQ_SYS_TRANS_HALF_TOPIC。
處理endTransaction方法
在endTransaction方法中將消息同步給Broker處理的Code對(duì)應(yīng)為END_TRANSACTION，Broker就會(huì)找出對(duì)應(yīng)的Processor來(lái)處理該類(lèi)即調(diào)用EndTransactionProcessor類(lèi)的processRequest方法處理。
if (MessageSysFlag.TRANSACTION_COMMIT_TYPE == requestHeader.getCommitOrRollback()) {
// 根據(jù)commitLogOffset獲取文件中的message，獲取到了返回success
result = this.brokerController.getTransactionalMessageService().commitMessage(requestHeader);
if (result.getResponseCode() == ResponseCode.SUCCESS) {
// 檢查消息是否一致
RemotingCommand res = checkPrepareMessage(result.getPrepareMessage(), requestHeader);
if (res.getCode() == ResponseCode.SUCCESS) {
// 生成要保存的消息
MessageExtBrokerInner msgInner = endMessageTransaction(result.getPrepareMessage());
msgInner.setSysFlag(MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), requestHeader.getCommitOrRollback()));
msgInner.setQueueOffset(requestHeader.getTranStateTableOffset());
msgInner.setPreparedTransactionOffset(requestHeader.getCommitLogOffset());
msgInner.setStoreTimestamp(result.getPrepareMessage().getStoreTimestamp());
MessageAccessor.clearProperty(msgInner, MessageConst.PROPERTY_TRANSACTION_PREPARED);
// 把真實(shí)的topic消息存儲(chǔ)到CommitLog中
RemotingCommand sendResult = sendFinalMessage(msgInner);
if (sendResult.getCode() == ResponseCode.SUCCESS) {
// 移除prepare消息，存入opQueueMap中
this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage());
}
return sendResult;
}
return res;
}
// 回滾
} else if (MessageSysFlag.TRANSACTION_ROLLBACK_TYPE == requestHeader.getCommitOrRollback()) {
// 查詢(xún)到half消息則返回成功
result = this.brokerController.getTransactionalMessageService().rollbackMessage(requestHeader);
if (result.getResponseCode() == ResponseCode.SUCCESS) {
// 檢查消息是否一致
RemotingCommand res = checkPrepareMessage(result.getPrepareMessage(), requestHeader);
if (res.getCode() == ResponseCode.SUCCESS) {
// 移除prepare消息，存入opQueueMap中
this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage());
}
return res;
}
}僅僅展示相關(guān)核心代碼，其主要邏輯：首先去判斷請(qǐng)求的方式是commit還是rollback，如果是commit查詢(xún)到消息還原消息原來(lái)的topic，然后刪除half topic上的消息轉(zhuǎn)存到opQueueMap中，如果是rollback直接進(jìn)行刪除half topic上的消息并轉(zhuǎn)存到opQueueMap中去。
注意：opQueueMap的引入為了解決有可能出現(xiàn)網(wǎng)絡(luò)、進(jìn)程、線(xiàn)程等各種因素導(dǎo)致消費(fèi)端未能成功處理消息的情況，該機(jī)制的作用是在消費(fèi)者端將未成功處理的消息重新發(fā)送到服務(wù)端進(jìn)行重試，直到確認(rèn)消息已經(jīng)被成功處理或者達(dá)到最大重試次數(shù)后進(jìn)行回滾操作。而 Op 消息本身則是通過(guò)修改消息狀態(tài)來(lái)實(shí)現(xiàn)的。
消息回查
當(dāng)網(wǎng)絡(luò)中斷或者響應(yīng)超時(shí)等各種異常信息導(dǎo)致消息并沒(méi)有傳送到broker端去，為了解決這一問(wèn)題在Broker就開(kāi)啟一個(gè)回查線(xiàn)程每隔一分鐘執(zhí)行一次處理超過(guò)6s未回查的消息，當(dāng)超過(guò)15次回查后直接將消息丟棄。
在啟動(dòng)BrokerController類(lèi)時(shí)，會(huì)去調(diào)用startProcessorByHa方法如果是Master節(jié)點(diǎn)就會(huì)去啟動(dòng)一個(gè)線(xiàn)程每隔6s處理未回查的消息，檢查最大次數(shù)為15次。
public void run() {
log.info("Start transaction check service thread!");
long checkInterval = brokerController.getBrokerConfig().getTransactionCheckInterval();
while (!this.isStopped()) {
this.waitForRunning(checkInterval);
}
log.info("End transaction check service thread!");
}
protected void onWaitEnd() {
long timeout = brokerController.getBrokerConfig().getTransactionTimeOut();
int checkMax = brokerController.getBrokerConfig().getTransactionCheckMax();
long begin = System.currentTimeMillis();
log.info("Begin to check prepare message, begin time:{}", begin);
// 檢查回查消息 timeout = 6s checkMax=15
this.brokerController.getTransactionalMessageService().check(timeout, checkMax, this.brokerController.getTransactionalMessageCheckListener());
log.info("End to check prepare message, consumed time:{}", System.currentTimeMillis() - begin);
}在check方法里面去調(diào)用listener.resolveHalfMsg(msgExt)方法去處理事務(wù)消息。
public void resolveHalfMsg(final MessageExt msgExt) {
executorService.execute(new Runnable() {
@Override
public void run() {
try {
sendCheckMessage(msgExt);
} catch (Exception e) {
LOGGER.error("Send check message error!", e);
}
}
});
}執(zhí)行sendCheckMessage方法發(fā)送一個(gè)檢查事務(wù)狀態(tài)的Code為CHECK_TRANSACTION_STATE的消息，在客戶(hù)端MQClientAPIImpl初始化的時(shí)候就會(huì)去注冊(cè)一個(gè)Code對(duì)應(yīng)的Processor，最終就會(huì)去執(zhí)行checkTransactionState方法，判斷本地事務(wù)的狀態(tài)，然后再去執(zhí)行endTransactionOneway發(fā)起END_TRANSACTION處理。
public void checkTransactionState(final String addr, final MessageExt msg,
final CheckTransactionStateRequestHeader header) {
Runnable request = new Runnable() {
private final String brokerAddr = addr;
private final MessageExt message = msg;
private final CheckTransactionStateRequestHeader checkRequestHeader = header;
private final String group = DefaultMQProducerImpl.this.defaultMQProducer.getProducerGroup();
// 執(zhí)行線(xiàn)程方法
@Override
public void run() {
TransactionCheckListener transactionCheckListener = DefaultMQProducerImpl.this.checkListener();
TransactionListener transactionListener = getCheckListener();
if (transactionCheckListener != null || transactionListener != null) {
LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
Throwable exception = null;
try {
if (transactionCheckListener != null) {
localTransactionState = transactionCheckListener.checkLocalTransactionState(message);
} else if (transactionListener != null) {
log.debug("Used new check API in transaction message");
// 檢查本地事務(wù)
localTransactionState = transactionListener.checkLocalTransaction(message);
} else {
log.warn("CheckTransactionState, pick transactionListener by group[{}] failed", group);
}
} catch (Throwable e) {
log.error("Broker call checkTransactionState, but checkLocalTransactionState exception", e);
exception = e;
}
// 處理事務(wù)狀態(tài)
this.processTransactionState(
localTransactionState,
group,
exception);
} else {
log.warn("CheckTransactionState, pick transactionCheckListener by group[{}] failed", group);
}
}
//
private void processTransactionState(
final LocalTransactionState localTransactionState,
final String producerGroup,
final Throwable exception) {
final EndTransactionRequestHeader thisHeader = new EndTransactionRequestHeader();
thisHeader.setCommitLogOffset(checkRequestHeader.getCommitLogOffset());
thisHeader.setProducerGroup(producerGroup);
thisHeader.setTranStateTableOffset(checkRequestHeader.getTranStateTableOffset());
thisHeader.setFromTransactionCheck(true);
thisHeader.setBname(checkRequestHeader.getBname());
String uniqueKey = message.getProperties().get(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
if (uniqueKey == null) {
uniqueKey = message.getMsgId();
}
thisHeader.setMsgId(uniqueKey);
thisHeader.setTransactionId(checkRequestHeader.getTransactionId());
switch (localTransactionState) {
// 提交狀態(tài)
case COMMIT_MESSAGE:
thisHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_COMMIT_TYPE);
break;
// 回滾狀態(tài)
case ROLLBACK_MESSAGE:
thisHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_ROLLBACK_TYPE);
log.warn("when broker check, client rollback this transaction, {}", thisHeader);
break;
// 未知狀態(tài)
case UNKNOW:
thisHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_NOT_TYPE);
log.warn("when broker check, client does not know this transaction state, {}", thisHeader);
break;
default:
break;
}
String remark = null;
if (exception != null) {
remark = "checkLocalTransactionState Exception: " + RemotingHelper.exceptionSimpleDesc(exception);
}
doExecuteEndTransactionHook(msg, uniqueKey, brokerAddr, localTransactionState, true);
try {
// 再次執(zhí)行endTransactionOneway發(fā)起END_TRANSACTION
DefaultMQProducerImpl.this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, thisHeader, remark,
3000);
} catch (Exception e) {
log.error("endTransactionOneway exception", e);
}
}
};
this.checkExecutor.submit(request);
}總結(jié)
首先客戶(hù)端Producer通過(guò)sendMessageInTransaction方法發(fā)送事務(wù)消息，Broker判斷是事務(wù)消息就將消息topic存入到RMQ_SYS_TRANS_HALF_TOPIC返回給客戶(hù)端，客戶(hù)端繼續(xù)執(zhí)行邏輯。
然后調(diào)用endTransaction方法去提交本地事務(wù)通過(guò)endTransactionOneway將消息提交給Broker端，Broker端通過(guò)Code為END_TRANSACTION的處理器去處理消息調(diào)用processRequest方法來(lái)處理對(duì)應(yīng)的消息，
如果由于各種原因?qū)е孪⒌氖鬏?，為了防止這些現(xiàn)象的出現(xiàn)所以在BrokerController啟動(dòng)時(shí)就啟動(dòng)一個(gè)線(xiàn)程每隔6s處理未回查的消息（檢查最大次數(shù)為15次）的任務(wù)來(lái)進(jìn)行消息的回查，簡(jiǎn)單來(lái)說(shuō)就是通過(guò)sendCheckMessage方法去注冊(cè)一個(gè)Code為CHECK_TRANSACTION_STATE的消息將內(nèi)容發(fā)送給客戶(hù)端，然后客戶(hù)端在啟動(dòng)時(shí)也注冊(cè)對(duì)應(yīng)Code的處理邏輯，通過(guò)processTransactionState方法去處理事務(wù)的狀態(tài)，如果正常最后還是會(huì)去執(zhí)行endTransactionOneway方法，完成事務(wù)消息。
