RocketMQ事务实现原理

CBeann

发布于 2023-12-25 19:20:00

1580

发布于 2023-12-25 19:20:00

1需求

RocketMQ一个优点是有事务特性，可以保证事务的最终一致性。举一个简单的例子，以电商为例，创建订单ID9527被创建后要保存到数据库，通过该订单通过MQ投递给其他系统进行消费。如果要保证订单数据入库与消息投递状态要保证最终一致性，要怎么做？

第一种情况，先保存数据库，在发送MQ

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，在保存数据库

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生产者

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会查看服务器中某个半消息主题中的消息，通过定时任务去看这个消息是是否发送成功，后面会介绍。

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注册事务回查监听器

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类型，设置了一个执行器和回查器，如下面代码所示

// 事务回查器
    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方法，在该方法里会给消息添加一个如下代码的属性，标记该消息为事务消息

MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");

此时消息会发送到broker。

总结：消息在producer会被拦截并且添加属性MessageConst.PROPERTY_TRANSACTION_PREPARED标记消息为事务消息

3.2broker收到消息源码分析

直接从SendMessageProcessor的asyncSendMessage方法看吧，该方法里会根据MessageConst.PROPERTY_TRANSACTION_PREPARED确定走哪个分支，如下图的transFlag（此时为“true”）属性

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方法，如下面代码所示，

public CompletableFuture<PutMessageResult> asyncPutHalfMessage(MessageExtBrokerInner messageInner) {
        return store.asyncPutMessage(parseHalfMessageInner(messageInner));
    }

很明显会parseHalfMessageInner方法对messageInner进行修改配置，这里重点是把主题改为一个默认的半消息主题

//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

//执行自定义的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中

//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。