分布式場景下的事務(wù)機制
事務(wù)消息是RocketMQ的一個非常特色的高級特性，它的基礎(chǔ)訴求是通過RocketMQ的事務(wù)機制，來保證上下游的數(shù)據(jù)?致性。
我們在單機版本下面只需要在業(yè)務(wù)方法上加上對應(yīng)的事務(wù)就可以達到效果，但是分布式的場景下，多個系統(tǒng)之間的協(xié)調(diào)配合，你無法知道到底是那個先執(zhí)行那個后執(zhí)行，當然在微服務(wù)里面存在Seate框架來保證事務(wù)，但是這事務(wù)的保證始終是心頭大患，只能用一句話形容魚和熊掌不可兼得。
而RocketMq的事務(wù)消息能夠在提升性能的情況下滿足要求，其主要實現(xiàn)是支持分布式情況下保障消息生產(chǎn)和本地事務(wù)的最終一致性，消息生產(chǎn)我們可以使用順序消息去執(zhí)行，這樣我們只需要滿足這兩個的事務(wù)即可。
實現(xiàn)過程
圖片
準備階段：生產(chǎn)者將消息發(fā)送到Broker，Broker向生產(chǎn)者發(fā)送ack表示消息發(fā)送成功，但是此時的消息為一個等待狀態(tài)，不會被消費者去消費。（生產(chǎn)者繼續(xù)執(zhí)行接下來的代碼）
確認階段：當我們執(zhí)行完所有的代碼后，本地事務(wù)要么回滾要么提交，此時當我們了解本地事務(wù)的狀態(tài)后，將結(jié)果推送給Broker做二次確認結(jié)果，如果為Commit則將修改激活準備推送給消費者，如果為Rollback則將消息進行回滾。
補償機制：當出現(xiàn)異常情況沒有發(fā)生二次確認，此時我們在固定時間后將會進行回查，檢查回查消息對應(yīng)的本地事務(wù)的狀態(tài)，重寫Commit或者Rollback。
涉及狀態(tài)以及注意點
事務(wù)消息存在三種狀態(tài)：
CommitTransaction：提交事務(wù)狀態(tài)，此狀態(tài)下允許消費者消費。
RollbackTransaction：回滾事務(wù)狀態(tài)，此狀態(tài)下消息會被刪除。
Unknown：中間狀態(tài)，此狀態(tài)下會等待本地事務(wù)處理結(jié)果進行對應(yīng)操作。
注意點：
本消息狀態(tài)是一種對消費者不可見的狀態(tài)，將消息的內(nèi)容放到系統(tǒng)Topic的RMQ_SYS_TRANS_HALF_TOPIC隊列里面去。
事務(wù)消息中的相關(guān)參數(shù)可以進行設(shè)置，比如：本地事務(wù)回查次數(shù)transactionCheckMax默認15次，本地事務(wù)回查的間隙transactionCheckInterval默認60s，超出后會直接將消息丟棄。
RocketMQ的事務(wù)消息是指應(yīng)用本地事務(wù)和發(fā)送消息操作可以定義到全局事務(wù)中，要么同時成功，要么同時失敗，通過RocketMQ的事務(wù)信息可以實現(xiàn)可靠消息的最終一致性方案。
源碼解析
Producer端通過構(gòu)建TransactionMQProducer對象綁定事務(wù)監(jiān)聽。
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方法來發(fā)送消息。
public TransactionSendResult sendMessageInTransaction(final Message msg,
final LocalTransactionExecuter localTransactionExecuter, final Object arg)
throws MQClientException {
// 檢查TransactionListener是否存在，如果不存在就直接拋異常
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ù)請求根據(jù)這個producergroup找到指定的channel
// 生產(chǎn)者能找到所有在同一個組的機器實例從而檢查事務(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對象執(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對象
EndTransactionRequestHeader requestHeader = new EndTransactionRequestHeader();
requestHeader.setTransactionId(transactionId);
// offset是prepare消息中offsetMsgId中獲取的
requestHeader.setCommitLogOffset(id.getOffset());
requestHeader.setBname(sendResult.getMessageQueue().getBrokerName());
// 社會提交/回滾狀態(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，通過endTransactionOneway方法創(chuàng)建Code為END_TRANSACTION的消息，然后在Broker就會找出對應(yīng)的Processor來處理。
Broker端處理
Broker總共存在兩個處理，首先針對第一個階段發(fā)送的Half消息，broker要進行相關(guān)的操作，后面endTransaction提交進來的事務(wù)狀態(tài)，針對三種狀態(tài)進行相關(guān)操作。
接收第一階段發(fā)送的Half消息
SendMessageProcessor的sendMessage方法中去執(zhí)行處理事務(wù)消息。
// 發(fā)送Half消息時，在屬性中設(shè)置了PROPERTY_TRANSACTION_PREPARED為true，這里根據(jù)這個屬性判斷是否是事務(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ù)消息進入這里，把消息的topic改成RMQ_SYS_TRANS_HALF_TOPIC，以同步刷盤的方式存入store
putMessageResult = this.brokerController.getTransactionalMessageService().prepareMessage(msgInner);
}如果消息攜帶事務(wù)標記就去執(zhí)行TransactionMessageService類的prepareMessage方法進行相關(guān)的處理。
// 解析Half消息
private MessageExtBrokerInner parseHalfMessageInner(MessageExtBrokerInner msgInner) {
// 把真實的topic和真實的queueId放在消息的屬性中
MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_TOPIC, msgInner.getTopic());
MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_QUEUE_ID,
String.valueOf(msgInner.getQueueId()));
// 設(shè)置默認的事務(wù)狀態(tài)為TRANSACTION_NOT_TYPE=>unknow
msgInner.setSysFlag(
MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), MessageSysFlag.TRANSACTION_NOT_TYPE));
// 將消息的topic設(shè)置為RMQ_SYS_TRANS_HALF_TOPIC，這個是對消費者不可見的
msgInner.setTopic(TransactionalMessageUtil.buildHalfTopic());
// 設(shè)置queueId=0
msgInner.setQueueId(0);
msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
return msgInner;
}進行topic的切換，將原來的topic存入到消息的屬性里面，將消息的topic設(shè)置為RMQ_SYS_TRANS_HALF_TOPIC。
處理endTransaction方法
在endTransaction方法中將消息同步給Broker處理的Code對應(yīng)為END_TRANSACTION，Broker就會找出對應(yīng)的Processor來處理該類即調(diào)用EndTransactionProcessor類的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);
// 把真實的topic消息存儲到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()) {
// 查詢到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)核心代碼，其主要邏輯：首先去判斷請求的方式是commit還是rollback，如果是commit查詢到消息還原消息原來的topic，然后刪除half topic上的消息轉(zhuǎn)存到opQueueMap中，如果是rollback直接進行刪除half topic上的消息并轉(zhuǎn)存到opQueueMap中去。
注意：opQueueMap的引入為了解決有可能出現(xiàn)網(wǎng)絡(luò)、進程、線程等各種因素導(dǎo)致消費端未能成功處理消息的情況，該機制的作用是在消費者端將未成功處理的消息重新發(fā)送到服務(wù)端進行重試，直到確認消息已經(jīng)被成功處理或者達到最大重試次數(shù)后進行回滾操作。而 Op 消息本身則是通過修改消息狀態(tài)來實現(xiàn)的。
消息回查
當網(wǎng)絡(luò)中斷或者響應(yīng)超時等各種異常信息導(dǎo)致消息并沒有傳送到broker端去，為了解決這一問題在Broker就開啟一個回查線程每隔一分鐘執(zhí)行一次處理超過6s未回查的消息，當超過15次回查后直接將消息丟棄。
在啟動BrokerController類時，會去調(diào)用startProcessorByHa方法如果是Master節(jié)點就會去啟動一個線程每隔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ā)送一個檢查事務(wù)狀態(tài)的Code為CHECK_TRANSACTION_STATE的消息，在客戶端MQClientAPIImpl初始化的時候就會去注冊一個Code對應(yīng)的Processor，最終就會去執(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í)行線程方法
@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é)
首先客戶端Producer通過sendMessageInTransaction方法發(fā)送事務(wù)消息，Broker判斷是事務(wù)消息就將消息topic存入到RMQ_SYS_TRANS_HALF_TOPIC返回給客戶端，客戶端繼續(xù)執(zhí)行邏輯。
然后調(diào)用endTransaction方法去提交本地事務(wù)通過endTransactionOneway將消息提交給Broker端，Broker端通過Code為END_TRANSACTION的處理器去處理消息調(diào)用processRequest方法來處理對應(yīng)的消息，
如果由于各種原因?qū)е孪⒌氖鬏?，為了防止這些現(xiàn)象的出現(xiàn)所以在BrokerController啟動時就啟動一個線程每隔6s處理未回查的消息（檢查最大次數(shù)為15次）的任務(wù)來進行消息的回查，簡單來說就是通過sendCheckMessage方法去注冊一個Code為CHECK_TRANSACTION_STATE的消息將內(nèi)容發(fā)送給客戶端，然后客戶端在啟動時也注冊對應(yīng)Code的處理邏輯，通過processTransactionState方法去處理事務(wù)的狀態(tài)，如果正常最后還是會去執(zhí)行endTransactionOneway方法，完成事務(wù)消息。
