1需求
RocketMQ一个优点是有事务特性，可以保证事务的最终一致性。举一个简单的例子，以电商为例，创建订单ID9527被创建后要保存到数据库，通过该订单通过MQ投递给其他系统进行消费。如果要保证订单数据入库与消息投递状态要保证最终一致性，要怎么做？
第一种情况，先保存数据库，在发送MQ
代码语言：javascript复制public void register(UserInfo userInfo){
//保存用户信息
userDao.insert(userInfo);
//发送MQ消息
SendResult res = mqProvider.sendMessage(userInfo.getId);
if(res.getCode().equals("SEND_OK")){
connection.commit();
}else{
connection.rollback();
}
}如果生产者发送消息时，因为网络原因导致10秒消息才返回SendResult结果，这就意味着10秒内数据库事务无法提交，高并发情况下，数据库连接池资源会被打满，后果是比较严重的。
第二种情况，先发送MQ，在保存数据库
代码语言：javascript复制public void register(UserInfo userInfo){
userDao.insert(userInfo);
//发送MQ消息
SendResult res = mqProvider.sendMessage(userInfo.getId);
if(res.getCode().equals("SEND_OK")){
//保存用户信息
userDao.insert(userInfo);
//此时系统宕机
connection.commit();
}else{
connection.rollback();
}
}如果消息发送成功，但是执行保存的时候数据库出现宕机，则MQ中却有了消息，但是数据库却没有该条订单信息，则MQ的下游消费方需要从业务逻辑上做额外处理。
问题很清晰了，我就是想要数据库保存和MQ发送同时成功或者同时失败，保证事务的最终一致性。下面分析RocketMQ是怎么实现的
2代码
首先我会发送消息内容为9527到broker,此时消息的状态是半消息状态（properties中一个标志位，即对消费不可见），并且存储到Commitlog中，把偏移量返回给producer,producer收到存储消息成功后，会调用TransactionExecuterimpl的executeLocalTransactionBranch方法，在该方法中执行保存数据库的操作，如果保存数据库成功，返回一个LocalTransactionState.COMMIT_MESSAGE，反之返回LocalTransactionState.ROLLBACK_MESSAGE，并且把返回值发送给broker,broker根据LocalTransactionState的状态确定是消息是否标记丢弃和标记成功，从而实现对用户是否可见。
如果在TransactionExecuterimpl的executeLocalTransactionBranch中数据库执行commit后系统宕机，此时数据库是有数据而borker的消息为半消息状态，这种情况下数据还是不一致，那么这时就用到事务回查器，broker会通过线程池扫描半消息的任务队列去发送请求给broker到TransactionCheckListenerImpl方法，比如查看数据库是否存在返回LocalTransactionState状态。
2.1生产者
代码语言：javascript复制package transaction;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
/**
* @author chaird
* @create 2021-10-25 11:53
*/
public class SyncProducer {
public static void main(String[] args) throws Exception {
// 事务回查器
TransactionCheckListenerImpl checkListener = new TransactionCheckListenerImpl();
// 事务消息生产者
TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");
// NameServer地址
producer.setNamesrvAddr("localhost:9876");
producer.setSendMsgTimeout(999999999);
// 注册事务回查监听
producer.setTransactionCheckListener(checkListener);
// 本地事务执行器
TransactionExecuterimpl executerimpl = new TransactionExecuterimpl();
producer.start();
// Create a message instance, specifying topic, tag and message body.
Message msg =
new Message(
"TopicTest" /* Topic */,
"TagA" /* Tag */,
("" 9527).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */);
// Call send message to deliver message to one of brokers.
SendResult sendResult = producer.sendMessageInTransaction(msg, executerimpl, null);
System.out.printf("%s%n", sendResult);
// Shut down once the producer instance is not longer in use.
producer.shutdown();
}
}2.2事务回查器
broker会查看服务器中某个半消息主题中的消息，通过定时任务去看这个消息是是否发送成功，后面会介绍。
代码语言：javascript复制package transaction;
/**
* @author chaird
* @create 2022-01-08 20:24
*/
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionCheckListener;
import org.apache.rocketmq.common.message.MessageExt;
public class TransactionCheckListenerImpl implements TransactionCheckListener {
@Override
public LocalTransactionState checkLocalTransactionState(MessageExt messageExt) {
System.out.println("服务器端回查事务消息： " messageExt.toString());
//由于RocketMQ迟迟没有收到消息的确认消息，因此主动询问这条prepare消息，是否正常？
//可以查询数据库看这条数据是否已经处理
return LocalTransactionState.COMMIT_MESSAGE;
}
}2.3注册事务回查监听器
代码语言：javascript复制package transaction;
/**
* @author chaird
* @create 2022-01-08 20:24
*/
import org.apache.rocketmq.client.producer.LocalTransactionExecuter;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.common.message.Message;
import java.util.Date;
public class TransactionExecuterimpl implements LocalTransactionExecuter {
@Override
public LocalTransactionState executeLocalTransactionBranch(
final Message message, final Object o) {
try {
String msg = new String(message.getBody(), "UTF-8");
System.out.println("TransactionExecuterimpl-------" msg);
// db.save()
// db.save()
// db.save()
// connect.commit()
System.out.println(new Date() "===> 本地事务执行成功，发送确认消息");
} catch (Exception e) {
System.out.println(new Date() "===> 本地事务执行失败,回滚！！！");
// connect.rollback()
return LocalTransactionState.ROLLBACK_MESSAGE;
}
return LocalTransactionState.COMMIT_MESSAGE;
}
}3部分源码分析
3.1客户端发送消息源码分析
客户端的生产者类型为TransactionMQProducer类型，设置了一个执行器和回查器，如下面代码所示
代码语言：javascript复制// 事务回查器
TransactionCheckListenerImpl checkListener = new TransactionCheckListenerImpl();
// 事务消息生产者
TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");
// 注册事务回查监听
producer.setTransactionCheckListener(checkListener);
// 本地事务执行器
TransactionExecuterimpl executerimpl = new TransactionExecuterimpl();
//创建消息
Message msg =
new Message(
"TopicTest" /* Topic */,
"TagA" /* Tag */,
("" i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */);
// 发送消息.
SendResult sendResult = producer.sendMessageInTransaction(msg, executerimpl, null);调用producer的sendMessageInTransaction方法，在该方法里会给消息添加一个如下代码的属性，标记该消息为事务消息
代码语言：javascript复制MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");此时消息会发送到broker。
总结：消息在producer会被拦截并且添加属性MessageConst.PROPERTY_TRANSACTION_PREPARED标记消息为事务消息
3.2broker收到消息源码分析
直接从SendMessageProcessor的asyncSendMessage方法看吧，该方法里会根据MessageConst.PROPERTY_TRANSACTION_PREPARED确定走哪个分支，如下图的transFlag（此时为“true”）属性
代码语言：javascript复制String transFlag = origProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);
if (transFlag != null && Boolean.parseBoolean(transFlag)) {
if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) {
//省略
}
putMessageResult = this.brokerController.getTransactionalMessageService().asyncPrepareMessage(msgInner);
} else {
putMessageResult = this.brokerController.getMessageStore().asyncPutMessage(msgInner);
}然后就跳转到TransactionalMessageBridge的asyncPutHalfMessage方法，如下面代码所示，
代码语言：javascript复制public CompletableFuture<PutMessageResult> asyncPutHalfMessage(MessageExtBrokerInner messageInner) {
return store.asyncPutMessage(parseHalfMessageInner(messageInner));
}很明显会parseHalfMessageInner方法对messageInner进行修改配置，这里重点是把主题改为一个默认的半消息主题
代码语言：javascript复制//public static final String RMQ_SYS_TRANS_HALF_TOPIC = "RMQ_SYS_TRANS_HALF_TOPIC";
msgInner.setTopic(TransactionalMessageUtil.buildHalfTopic());构建完毕后调用DefaultMessageStore的asyncPutMessage方法把数据方法CommitLog里，此时把偏移量返回给producer
总结：消息在broker会被换主题，发到一个特定的主题里，并且把消息在CommitLog的offset发送给proudcer
3.3客户端发送消息后收到broker消息处理本地事务并且反馈结果给broker
DefaultMQProducerImpl调用sendMessageInTransaction后发送消息并且拿到响应结果，如图是放入CommitLog成功，则会执行LocalTransactionExecuter（看上面的代码，逻辑为运行MySQL,并且跟进成功与否返回一个状态localTransactionState），然后执行DefaultMQProducerImpl的endTransaction方法，把localTransactionState状态发送给broker
代码语言：javascript复制//执行自定义的LocalTransactionExecuter，即上面的TransactionExecuterimpl
localTransactionState = localTransactionExecuter.executeLocalTransactionBranch(msg, arg);
this.endTransaction(msg, sendResult, localTransactionState, localException);总结：客户端确定消息（半消息状态）已经被放在CommitLog中，执行自定义的LocalTransactionExecuter（逻辑为根据数据库是否成功提交返回localTransactionState状态，即确定消息是否为丢弃或者修改为提交状态）
3.4broker收到producer的本地事务状态修改消息的状态
直接从EndTransactionProcessor的processRequest方法开始说吧
如果producer中返回的MessageSysFlag.TRANSACTION_COMMIT_TYPE，即本地事务执行成功。
首先会根据客户端发送的偏移量去CommitLog中取到消息,然后创建一个新的消息对象放入到CommitLog中
代码语言：javascript复制//EndTransactionProcessor###processRequest
if (MessageSysFlag.TRANSACTION_COMMIT_TYPE == requestHeader.getCommitOrRollback()) {
//根据prdducer中传过来的偏移量获取消息内容
//根据prdducer中传过来的偏移量获取消息内容
//根据prdducer中传过来的偏移量获取消息内容
result = this.brokerController.getTransactionalMessageService().commitMessage(requestHeader);
if (result.getResponseCode() == ResponseCode.SUCCESS) {
RemotingCommand res = checkPrepareMessage(result.getPrepareMessage(), requestHeader);
if (res.getCode() == ResponseCode.SUCCESS) {
//把取出来的消息内容重新copy一个新对象
//把取出来的消息内容重新copy一个新对象
//把取出来的消息内容重新copy一个新对象
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);
// 把新的数据在重新放在CommitLog中，生成ConsumeQueue
// 把新的数据在重新放在CommitLog中，生成ConsumeQueue
// 把新的数据在重新放在CommitLog中，生成ConsumeQueue
RemotingCommand sendResult = sendFinalMessage(msgInner);
if (sendResult.getCode() == ResponseCode.SUCCESS) {
this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage());
}
return sendResult;
}
return res;
}总结：把原来的消息（半消息主题）重新复制一份（更正为正确的主题）插入到CommitLog，构建CounsumeQueue。
4参考
彻底看懂RocketMQ事务实现原理 - 知乎
【IT老齐087】先写库还是先发消息？RocketMQ如何保证消息与事务一致性？_哔哩哔哩_bilibili
