如果同一个数据源在本地事物很好控制，但是在不断发展的互联网环境下，微服务越来越流行，这个时候，需要解决分布式事物，需要保证数据的最终一致性。
所谓分布式事务（全局事物），就是在两个不同的系统中（即两个不同的数据源，其实同一个数据源也行），无法同一个同一个spring事物去控制不同系统的事物的整体成功或者整体失败。
分布式事务的解决方法很多，但是性能和复杂性不一样，今天说说rocketmq如何保证分布式事务的。
rocketmq为我们提供TransactionMQProducer API的支持。这个api在发送消息的时候主要做了三件事：
1，先发送需要发送的消息到消息中间件broker，并获取到该message的transactionId。在第一次发送的时候，该消息的状态为LocalTransactionState.UNKNOW
2，处理本地事物。
3，根据本地事物的执行结果，结合transactionId，找到该消息的位置，在mq中标志该消息的最终处理结果。
整体的执行源码逻辑如下：
public TransactionSendResult sendMessageInTransaction(Message msg, LocalTransactionExecuter tranExecuter, Object arg) throws MQClientException {
if (null == tranExecuter) {
throw new MQClientException("tranExecutor is null", (Throwable)null);
} else {
Validators.checkMessage(msg, this.defaultMQProducer);
SendResult sendResult = null;
MessageAccessor.putProperty(msg, "TRAN_MSG", "true");
MessageAccessor.putProperty(msg, "PGROUP", this.defaultMQProducer.getProducerGroup());
try {
//这里执行第一次发送消息，也就是预发送，并获取sendResult，这里包含msg的所有消息
sendResult = this.send(msg);
} catch (Exception var10) {
throw new MQClientException("send message Exception", var10);
}
LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
Throwable localException = null;
//根据预发送消息的状态做不同的处理，这里主要看SEND_OK
switch(sendResult.getSendStatus()) {
case SEND_OK:
try {
if (sendResult.getTransactionId() != null) {
msg.putUserProperty("__transactionId__", sendResult.getTransactionId());
}
// 这里做第二步，执行业务逻辑，即本地事物，
//具体的本地事物在LocalTransactionExecuter参数的实现类中，
//需要根据自己的业务逻辑去写，下面的//tranExecuter.executeLocalTransactionBranch(msg, arg);会执行实
//现类中的executeLocalTransactionBranch业务。
localTransactionState = tranExecuter.executeLocalTransactionBranch(msg, arg);
if (null == localTransactionState) {
localTransactionState = LocalTransactionState.UNKNOW;
}
if (localTransactionState != LocalTransactionState.COMMIT_MESSAGE) {
this.log.info("executeLocalTransactionBranch return {}", localTransactionState);
this.log.info(msg.toString());
}
} catch (Throwable var9) {
this.log.info("executeLocalTransactionBranch exception", var9);
this.log.info(msg.toString());
localException = var9;
}
break;
case FLUSH_DISK_TIMEOUT:
case FLUSH_SLAVE_TIMEOUT:
case SLAVE_NOT_AVAILABLE:
localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
}
try {
// 这里的方法，其中的localTransactionState是第二次执行业务逻辑的结果
//可以根据这个结果，知道本地事物执行的成功还是失败。或者是异常localException，
//这样可以根据第一次发送消息的结果sendResult，去修改mq中第一次发送消息的状态，完成第三步操作。
this.endTransaction(sendResult, localTransactionState, localException);
} catch (Exception var8) {
this.log.warn("local transaction execute " + localTransactionState + ", but end broker transaction failed", var8);
}
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;
}
}
上述第三步执行的方法endTransaction，逻辑如下：
public void endTransaction(SendResult sendResult, LocalTransactionState localTransactionState, Throwable localException) throws RemotingException, MQBrokerException, InterruptedException, UnknownHostException {
// 获取第一次发送消息的id
MessageId id;
if (sendResult.getOffsetMsgId() != null) {
id = MessageDecoder.decodeMessageId(sendResult.getOffsetMsgId());
} else {
id = MessageDecoder.decodeMessageId(sendResult.getMsgId());
}
//获取事物id
String transactionId = sendResult.getTransactionId();
String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(sendResult.getMessageQueue().getBrokerName());
EndTransactionRequestHeader requestHeader = new EndTransactionRequestHeader();
requestHeader.setTransactionId(transactionId);
requestHeader.setCommitLogOffset(id.getOffset());
//根据本地事物执行状态localTransactionState，告知mq修改状态
switch(localTransactionState) {
case COMMIT_MESSAGE:
requestHeader.setCommitOrRollback(Integer.valueOf(8));
break;
case ROLLBACK_MESSAGE:
requestHeader.setCommitOrRollback(Integer.valueOf(12));
break;
case UNKNOW:
requestHeader.setCommitOrRollback(Integer.valueOf(0));
}
requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
requestHeader.setTranStateTableOffset(sendResult.getQueueOffset());
requestHeader.setMsgId(sendResult.getMsgId());
String remark = localException != null ? "executeLocalTransactionBranch exception: " + localException.toString() : null;
//具体执行第三步完成整个事务。
this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, requestHeader, remark, (long)this.defaultMQProducer.getSendMsgTimeout());
}
上述：
如果第三阶段出现异常或者网络原因，就是本地事务执行成功持久化到数据库中，但是在修改mq中消息状态出现异常的时候，这样就可以出现本地和mq的消息状态的不一致问题。或者说，所有的数据不一致问题。rocketmq都会定期通过TransactionMQProducer API初始化的时候，设置的TransactionCheckListener的的实现类的checkLocalTransactionState 方法检查本地消息的状态，根据本地状态修改mq的状态
package org.apache.rocketmq.client.producer;
import org.apache.rocketmq.common.message.MessageExt;
public interface TransactionCheckListener {
LocalTransactionState checkLocalTransactionState(MessageExt var1);
}
这样就可以确保生产端的数据要么是本地事物执行失败，rollback mq中消息，或者本地事物执行成功，mq中的消息也是待消费的状态。
说到这里，其实分布式事务已经完成90%。如果生产端的的第三部不是成功状态，消费端是不会消费第一阶段的消息的。所以，只要消费端消费的消息，肯定是生产端成功的消息，而消费端只要保证幂等性，就可以准确的消费消息（因为消费失败了，mq是可以重试，如果最总都没有被消费，基本是程序有bug，可以事后人工处理了。）
网友评论