RocketMQ事务消息源码解析及关键点总结(基于4.0.0+版本)

柔光的暖阳◎ 2022-05-28 09:16 124阅读 0赞

RocketMQ在3.1.2以前有事务消息回查机制，也就是Broker对CommitLog内长时间处于Prepared状态的事务消息，向发送器消息的Producer发送请求，询问当前消息是提交还是回滚，然后根据结果去修改此消息的状态。由于涉及到了数据的修改，内存中存在很多修改了数据的高速缓存页，也就是脏页，很影响内存的使用效率，所以RocketMQ在3.1.2以后就移除了Broker的消息回查机制。

因为去除了消息回查，也移除了消息状态的修改，其事务消息的实现机制也做了一些调整，具体流程下面开始讲。

先将生产者发送事务消息，生产者需要使用TransactionMQProducer对象来发送消息，看起源码：

public class TransactionMQProducer extends DefaultMQProducer { 
        private TransactionCheckListener transactionCheckListener;
        private int checkThreadPoolMinSize = 1;
        private int checkThreadPoolMaxSize = 1;
        private int checkRequestHoldMax = 2000;
    
        public TransactionMQProducer() {
        }
    
        public TransactionMQProducer(final String producerGroup) {
            super(producerGroup);
        }
    
        public TransactionMQProducer(final String producerGroup, RPCHook rpcHook) {
            super(producerGroup, rpcHook);
        }
    
        @Override
        public void start() throws MQClientException {
            this.defaultMQProducerImpl.initTransactionEnv();
            super.start();
        }
    
        @Override
        public void shutdown() {
            super.shutdown();
            this.defaultMQProducerImpl.destroyTransactionEnv();
        }
    
        @Override
        public TransactionSendResult sendMessageInTransaction(final Message msg,
            final LocalTransactionExecuter tranExecuter, final Object arg) throws MQClientException {
            if (null == this.transactionCheckListener) {
                throw new MQClientException("localTransactionBranchCheckListener is null", null);
            }
    
            return this.defaultMQProducerImpl.sendMessageInTransaction(msg, tranExecuter, arg);
        }
    
        ......
    }

TransactionMQProducer 继承自DefaultMQProducer ，所以生产者的配置和默认的生产者一样。在start时，相比于DefaultMQProducer 多了步执行initTransactionEnv( )方法：

public void initTransactionEnv() {
        TransactionMQProducer producer = (TransactionMQProducer) this.defaultMQProducer;
        this.checkRequestQueue = new LinkedBlockingQueue<Runnable>(producer.getCheckRequestHoldMax());
        this.checkExecutor = new ThreadPoolExecutor(
            producer.getCheckThreadPoolMinSize(),
            producer.getCheckThreadPoolMaxSize(),
            1000 * 60,
            TimeUnit.MILLISECONDS,
            this.checkRequestQueue);
    }

初始化消息回查线程池，当有消息回查时，将任务线程放入线程池中，不过这版本不会产生作用。

TransactionMQProducer 在发送消息前需要检查transactionCheckListener属性，也就是消息回查时的结果返回：

/** * 【事务消息回查】检查监听器 */
    public interface TransactionCheckListener { 
    
        /** * 获取（检查）【本地事务】状态 * * @param msg 消息 * @return 事务状态 */
        LocalTransactionState checkLocalTransactionState(final MessageExt msg);
    
    }

虽然当前版本移除了，但是发送事务消息时还是需要指定的，对这个接口定义一个实现了，注入的事务消息生产者中，生产者定义形式可参照如下：

@Bean(initMethod = "start")
    public TransactionMQProducer transactionMQProducer() throws MQClientException {
        TransactionMQProducer producer = new TransactionMQProducer("tx_group");
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.setInstanceName("192.168.0.1@110");
        // 必须设为false否则连接broker10909端口
        producer.setVipChannelEnabled(false);
        producer.setTransactionCheckListener(new DefaultTransactionCheckListener());
        return producer;
    }

下面看发送事务消息的关键方法：

public class DefaultMQProducerImpl implements MQProducerInner { 
    
        /** * 发送事务消息 * * @param msg 消息 * @param tranExecuter 【本地事务】执行器 * @param arg 【本地事务】执行器参数 * @return 事务发送结果 * @throws MQClientException 当 Client 发生异常时 */
        public TransactionSendResult sendMessageInTransaction(final Message msg, final LocalTransactionExecuter tranExecuter, final Object arg)
            throws MQClientException {
            if (null == tranExecuter) {
                throw new MQClientException("tranExecutor is null", null);
            }
            Validators.checkMessage(msg, this.defaultMQProducer);
    
            // 发送【Half消息】
            SendResult sendResult;
            MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
            MessageAccessor.putProperty(msg, MessageConst.PROPERTY_PRODUCER_GROUP, this.defaultMQProducer.getProducerGroup());
            try {
                sendResult = this.send(msg);
            } catch (Exception e) {
                throw new MQClientException("send message Exception", e);
            }
    
            // 处理发送【Half消息】结果
            LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
            Throwable localException = null;
            switch (sendResult.getSendStatus()) {
                // 发送【Half消息】成功，执行【本地事务】逻辑
                case SEND_OK: {
                    try {
                        if (sendResult.getTransactionId() != null) { // 事务编号。目前开源版本暂时没用到，猜想ONS在使用。
                            msg.putUserProperty("__transactionId__", sendResult.getTransactionId());
                        }
    
                        // 执行【本地事务】逻辑
                        localTransactionState = tranExecuter.executeLocalTransactionBranch(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;
                // 发送【Half消息】失败，标记【本地事务】状态为回滚
                case FLUSH_DISK_TIMEOUT:
                case FLUSH_SLAVE_TIMEOUT:
                case SLAVE_NOT_AVAILABLE:
                    localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
                    break;
                default:
                    break;
            }
    
            // 结束事务：提交消息 COMMIT / ROLLBACK
            try {
                this.endTransaction(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;
        }
    
    }

向Message的Properties中添加 MessageConst.PROPERTY\_ TRANSACTION\_PREPARED属性，其值 “true”。  
在发送消息的通用方法中，获取此属性，当其值为 “true”时，对消息的系统标识加上 MessageSysFlag.TRANSACTION\_ PREPARED\_TYPE：

public class DefaultMQProducerImpl implements MQProducerInner { 
    
        private SendResult sendKernelImpl(......) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
            ......
            final String tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
            if (tranMsg != null && Boolean.parseBoolean(tranMsg)) {
                sysFlag |= MessageSysFlag.TRANSACTION_PREPARED_TYPE; 
            }
            ......
    
    }

将此消息添加事务的准备标识，在Broker接受到此消息时会做如下处理：

/** * 写入消息到Buffer默认实现 */
    class DefaultAppendMessageCallback implements AppendMessageCallback { 
    
        //向CommitLog追加消息
        public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank, final MessageExtBrokerInner msgInner) {
            ......
            //记录这条消息的消费信息在当前队列的序号,也就是第几条消息
            Long queueOffset = CommitLog.this.topicQueueTable.get(key);
            if (null == queueOffset) {
                queueOffset = 0L;
                CommitLog.this.topicQueueTable.put(key, queueOffset);
            }
            // Transaction messages that require special handling
            final int tranType = MessageSysFlag.getTransactionValue(msgInner.getSysFlag());
            switch (tranType) {
                // Prepared and Rollback message is not consumed, will not enter the
                // consumer queue
                case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
                case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
                    queueOffset = 0L;
                    break;
                case MessageSysFlag.TRANSACTION_NOT_TYPE:
                case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
                default:
                    break;
            }
            ......
            switch (tranType) {
                //事务类型为PREPARED_TYPE和ROLLBACK_TYPE的消息不会进ConsumeQueue,所以这里不会增加queueOffset的值
                case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
                case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
                    break;
                case MessageSysFlag.TRANSACTION_NOT_TYPE:
                case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
                    // The next update ConsumeQueue information 更新队列的offset
                    CommitLog.this.topicQueueTable.put(key, ++queueOffset);
                    break;
                default:
                    break;
            }
    
    }

提取消息的系统标识，当其为PREPARED\_ TYPE 和 ROLLBACK\_TYPE类型的消息时，设置其queueOffset为0，同时在存储这两种类型的消息时，ConsumeQueue的Offset不会自增，为何如此，看如下代码：

/** * 重放消息线程服务 * 该服务不断生成 消息位置信息 到 消费队列(ConsumeQueue) * 该服务不断生成 消息索引 到 索引文件(IndexFile) */
    class ReputMessageService extends ServiceThread {
    
        private void doReput() {
            ......
            // 成功读取Message,更新ConsumeQueue里的位置信息,更新IndexFile
            DefaultMessageStore.this.doDispatch(dispatchRequest);
            ......
        }
    
    }
    
    
    public class DefaultMessageStore implements MessageStore { 
    
        /** * 执行调度请求 * 1. 非事务消息 或 事务提交消息 建立 消息位置信息 到 ConsumeQueue * 2. 建立 索引信息 到 IndexFile * * @param req 调度请求 */
        public void doDispatch(DispatchRequest req) {
            // 非事务消息 或 事务提交消息 建立 消息位置信息 到 ConsumeQueue
            final int tranType = MessageSysFlag.getTransactionValue(req.getSysFlag());
            switch (tranType) {
                case MessageSysFlag.TRANSACTION_NOT_TYPE: // 非事务消息
                case MessageSysFlag.TRANSACTION_COMMIT_TYPE: // 事务消息COMMIT
                    DefaultMessageStore.this.putMessagePositionInfo(req.getTopic(), req.getQueueId(), req.getCommitLogOffset(), req.getMsgSize(),
                        req.getTagsCode(), req.getStoreTimestamp(), req.getConsumeQueueOffset());
                    break;
                case MessageSysFlag.TRANSACTION_PREPARED_TYPE: // 事务消息PREPARED
                case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE: // 事务消息ROLLBACK
                    break;
            }
            // 建立 索引信息 到 IndexFile,不支持对ROLLBACK类型的消息建立索引信息。
            if (DefaultMessageStore.this.getMessageStoreConfig().isMessageIndexEnable()) {
                DefaultMessageStore.this.indexService.buildIndex(req);
            }
        }
    
    }

ReputMessageService在执行消息重放时，如果消息类型是PREPARED和ROLLBACK，那么会忽略生成MessagePositionInfo，也就是其位置信息不会存放入ConsumeQueue，所以CommitLog在存储消息时，直接设置其consumeOffset = 0，且对应的ConsumeQueue的Offset不自增。

**PREPARED和ROLLBACK类型的消息位置信息不会存入ConsumeQueue，而消费者是通过ConsumeQueue获取消息的CommitLog Offset后再去CommitLog里提取消息的，也就是说PREPARED和ROLLBACK类型的消息不会被消费，这两种类型的消息也不支持延时投递**

在向Broker发型Half消息成功后，接下来会执行本地事务逻辑，此时会用到一个LocalTransactionExecuter对象，看起源码：

/** * 本地事务执行器 */
    public interface LocalTransactionExecuter { 
    
        /** * 执行本地事务分支 * * @param msg * @param arg * @return */
        LocalTransactionState executeLocalTransactionBranch(final Message msg, final Object arg);
    }
    
    public enum LocalTransactionState {
        COMMIT_MESSAGE,
        ROLLBACK_MESSAGE,
        UNKNOW,
    }

也就是说先向Broker发送一个Prepared消息，在发送成功后执行本地逻辑，本地逻辑可返回COMMIT\_ MESSAGE，或者ROLLBACK\_MESSAGE，根据枚举常量的值来判断是提交还是回滚消息：

public class DefaultMQProducerImpl implements MQProducerInner { 
    
        /** * 结束事务：提交消息 COMMIT / ROLLBACK * * @param sendResult 发送【Half消息】结果 * @param localTransactionState 【本地事务】状态 * @param localException 执行【本地事务】逻辑产生的异常 * @throws RemotingException 当远程调用发生异常时 * @throws MQBrokerException 当 Broker 发生异常时 * @throws InterruptedException 当线程中断时 * @throws UnknownHostException 当解码消息编号失败是 */
        public void endTransaction(final SendResult sendResult,
                                   final LocalTransactionState localTransactionState,
                                   final Throwable localException) throws RemotingException, MQBrokerException, InterruptedException, UnknownHostException {
            // 解码消息编号
            final MessageId id;
            if (sendResult.getOffsetMsgId() != null) {
                id = MessageDecoder.decodeMessageId(sendResult.getOffsetMsgId());
            } else {
                id = MessageDecoder.decodeMessageId(sendResult.getMsgId());
            }
    
            // 创建请求
            String transactionId = sendResult.getTransactionId();
            final String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(sendResult.getMessageQueue().getBrokerName());
            EndTransactionRequestHeader requestHeader = new EndTransactionRequestHeader();
            requestHeader.setTransactionId(transactionId);
            requestHeader.setCommitLogOffset(id.getOffset());
            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;
            }
            requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
            requestHeader.setTranStateTableOffset(sendResult.getQueueOffset());
            requestHeader.setMsgId(sendResult.getMsgId());
            String remark = localException != null ? ("executeLocalTransactionBranch exception: " + localException.toString()) : null;
    
            // 提交消息 COMMIT / ROLLBACK。！！！通信方式为：Oneway！！！
            this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, requestHeader, remark, this.defaultMQProducer.getSendMsgTimeout());
        }
    
    }

向之前接收到Prepared消息的Broker发送消息结果，带上本地逻辑的执行结果，Prepared消息的CommitLog offset，通信方式为Oneway，也就是不等待返回结果。发送过程可能因为网络或者其他原因出现发送失败，原先有Broker的回查功能，但现版本已经移除，所以需要自行定义类似的逻辑。

下面看Broker收到事务确认消息的处理过程：

/** * 事务结束处理器,只处理Commit/Rollback动作 */
    public class EndTransactionProcessor implements NettyRequestProcessor { 
    
        /** * 结束事务，提交/回滚消息 * * @param ctx ctx * @param request 请求 * @return 响应 * @throws RemotingCommandException 当解析请求失败时 */
        @Override
        public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
            ......
            // 将Prepared的消息从CommitLog查询出来
            final MessageExt msgExt = this.brokerController.getMessageStore().lookMessageByOffset(requestHeader.getCommitLogOffset());
            ......
            // 提取之前存储的事务消息数据,清除延时级别
            MessageExtBrokerInner msgInner = this.endMessageTransaction(msgExt);
            //设置事务消息的处理结果
            msgInner.setSysFlag(MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), requestHeader.getCommitOrRollback()));
            ......
            //如果处理结果是ROLLBACK,清除消息体
            if (MessageSysFlag.TRANSACTION_ROLLBACK_TYPE == requestHeader.getCommitOrRollback()) {
                msgInner.setBody(null);
            }
    
            // 存储事务结果消息
            final MessageStore messageStore = this.brokerController.getMessageStore();
            final PutMessageResult putMessageResult = messageStore.putMessage(msgInner);
        }
    
        /** * 生成事务结果消息 * * @param msgExt 消息 * @return 消息 */
        private MessageExtBrokerInner endMessageTransaction(MessageExt msgExt) {
            MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
            //确认消息(Rollback/Commit)和Prepared消息的内容一致
            msgInner.setBody(msgExt.getBody());
            msgInner.setFlag(msgExt.getFlag());
            MessageAccessor.setProperties(msgInner, msgExt.getProperties());
    
            TopicFilterType topicFilterType =
                (msgInner.getSysFlag() & MessageSysFlag.MULTI_TAGS_FLAG) == MessageSysFlag.MULTI_TAGS_FLAG ? TopicFilterType.MULTI_TAG
                    : TopicFilterType.SINGLE_TAG;
            long tagsCodeValue = MessageExtBrokerInner.tagsString2tagsCode(topicFilterType, msgInner.getTags());
            msgInner.setTagsCode(tagsCodeValue);
            msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgExt.getProperties()));
    
            msgInner.setSysFlag(msgExt.getSysFlag());
            msgInner.setBornTimestamp(msgExt.getBornTimestamp());
            msgInner.setBornHost(msgExt.getBornHost());
            msgInner.setStoreHost(msgExt.getStoreHost());
            msgInner.setReconsumeTimes(msgExt.getReconsumeTimes());
    
            msgInner.setWaitStoreMsgOK(false);  //不等待Slave是的进度上报
            MessageAccessor.clearProperty(msgInner, MessageConst.PROPERTY_DELAY_TIME_LEVEL); // 清除延迟级别=》事务消息不支持延时投递
    
            msgInner.setTopic(msgExt.getTopic());
            msgInner.setQueueId(msgExt.getQueueId());
    
            return msgInner;
        }
    
    }

通过以上代码可以看出，提交事务处理结果消息时，会重新生成一条消息，此消息以之前Prepared的消息为模板，很多数据都是直接拷贝，特别是body，也就是消息内容。同时设置消息处理结果，当结果为Rollback时，设置body为null，因为此类型的消息不会被消费，所以消息体每必要存储了，浪费空间。

**当消息的处理结果为Commit时，克隆Prepared消息的大部分信息，特别是body，然后再次存储一遍**。Commit类型的事务消息的PositionInfo会存入ConsumeQueue，也会正常生成索引信息存入IndexFile，能被Consumer正常查询和消费。