likes
comments
collection
share

RocketMQ源码分析13：事务消息

作者站长头像
站长
· 阅读数 42

RocketMQ源码分析13：事务消息

基于rocketmq-4.9.0 版本分析rocketmq 前面我们分析了RocketMQ的延迟消息,接下来我们看下RocketMQ的事务消息,参考官方文档

RocketMQ在4.3.0版中开始支持分布式事务消息，RocketMQ采用了2PC的思想来实现了提交事务消息，同时增加一个补偿逻辑来处理二阶段超时或者失败的消息，如下图所示。

RocketMQ源码分析13：事务消息

1.RocketMQ事务消息流程概要

上图说明了事务消息的大致方案，其中分为两个流程：正常事务消息的发送及提交、事务消息的补偿流程。 1.事务消息发送及提交： (1) 发送消息（这个消息暂时称为：half消息(半事务消息)) (2) 服务端响应消息写入结果。 (3) 根据发送结果执行本地事务（如果写入失败，此时half消息对业务不可见，本地逻辑不执行） (4) 根据本地事务状态执行Commit或者Rollback（Commit操作生成消息索引，消息对消费者可见）

2.补偿流程： (1) 对没有Commit/Rollback的事务消息（pending状态的消息），从服务端发起一次“回查” (2) Producer收到回查消息，检查回查消息对应的本地事务的状态 (3) 根据本地事务状态，重新Commit或者Rollback

其中，补偿阶段用于解决消息Commit或者Rollback发生超时或者失败的情况。

2. RocketMQ事务消息设计

在分析源码之前，我们先补充一些概念

2.1 分布式事务

对于分布式事务，通俗地说就是，一次操作由若干分支操作组成，这些分支操作分属不同应用，分布在不同服务器上。分布式事务需要保证这些分支操作要么全部成功，要么全部失败。分布式事务与普通事务一样，就是为了保证操作结果的一致性。

与之对应的就是本地事务：本地事务更多的是通过关系型数据库来控制事务，这是利用数据库本身的事务特性来实现的，因此叫数据库事务。数据库事务具有ACID特性，在实现时会将一次事务的所有操作全部纳入到一个不可分割的执行单元，该执行单元的所有操作要么都成功，要么都失败，只要其中任一操作执行失败，都将导致整个事务的回滚。

2.2 事务消息

RocketMQ提供了类似X/Open XA的分布式事务功能，通过事务消息能达到分布式事务的最终一致。XA是一种分布式事务解决方案，一种分布式事务处理模式。

2.3 半事务消息

暂不能消费的消息，Producer已经成功地将消息发送到了Broker，但是Broker未收到最终确认指令，此时该消息被标记成“暂不能消费”状态，即不能被消费者看到。处于该种状态下的消息即半事务消息。

2.4 消息回查

消息回查，即重新查询本地事务的执行状态（也就是上图中的步骤6）。假如张三给李四转账，张三首先发送一条需要转账的消息到MQ, 发送成功后，张三开始从建设银行卡扣款，可能由于网络的问题，张三在扣款的时候发生了故障，出现了扣款未知(UNKNOW, 就不是成功也不是失败)的状态，紧接着张三将这条UNKNOW状态的消息发给MQ, MQ 接收到是UNKNOW状态，则需要发起回查，给张三个机会，看看网络是不是好了，好了就扣款，如果是余额不足，那就结束转账。在RocketMQ中，事务消息有三个状态：

public enum LocalTransactionState {
    //TODO:本地事务执行成功，给broker发送一个commit的标识
    COMMIT_MESSAGE,
    
    ROLLBACK_MESSAGE,
    
    //TODO: 这个状态将会引起回查
    UNKNOW,
}

其中 LocalTransactionState.UNKNOW 状态，将会引起回查。

2.5 XA协议

XA（Unix Transaction）是一种分布式事务解决方案，一种分布式事务处理模式，是基于XA协议的。XA协议由Tuxedo（Transaction for Unix has been Extended for Distributed Operation，分布式操作扩展之后的Unix事务系统）首先提出的，并交给X/Open组织，作为资源管理器与事务管理器的接口标准.

XA模式中有三个重要组件：TC、TM、RM

  • TC :Transaction Coordinator，事务协调者。维护全局和分支事务的状态，驱动全局事务提交或回滚。

RocketMQ中Broker充当着TC。

  • TM : Transaction Manager，事务管理器。定义全局事务的范围：开始全局事务、提交或回滚全局事务。它实际是全局事务的发起者。

RocketMQ中事务消息的Producer充当着TM。

  • RM : Resource Manager，资源管理器。管理分支事务处理的资源，与TC交谈以注册分支事务和报告分支事务的状态，并驱动分支事务提交或回滚。

RocketMQ中事务消息的Producer及Broker均是RM

XA模式是一个典型的2PC，其执行原理如下:

  1. TM向TC发起指令，开启一个全局事务。
  2. 根据业务要求，各个RM会逐个向TC注册分支事务，然后TC会逐个向RM发出预执行指令。
  3. 各个RM在接收到指令后会在进行本地事务预执行。
  4. RM将预执行结果上报给TC。当然，这个结果可能是成功，也可能是失败。
  5. TC在接收到各个RM的Report后会将汇总结果上报给TM，根据汇总结果TM会向TC发出确认指令。TC在接收到指令后再次向RM发送确认指令。
  1. 若所有结果都是成功响应，则向TC发送Global Commit指令。
  2. 只要有结果是失败响应，则向TC发送Global Rollback指令。
  1. TC在接收到指令后再次向RM发送确认指令。

了解这些基本概念后，我们在回过头来看RocketMQ的事务消息的设计

  1. 事务消息在一阶段对用户(消费者)不可见 在RocketMQ事务消息的主要流程中，一阶段的消息如何对用户不可见。其中，事务消息相对普通消息最大的特点就是一阶段发送的消息对用户是不可见的。那么，如何做到写入消息但是对用户不可见呢？RocketMQ事务消息的做法是：如果消息是half消息，将备份原消息的主题与消息消费队列，然后改变主题为RMQ_SYS_TRANS_HALF_TOPIC。由于消费组未订阅该主题，故消费端无法消费half类型的消息，然后RocketMQ会开启一个定时任务，从Topic为RMQ_SYS_TRANS_HALF_TOPIC中拉取消息进行消费，根据生产者组获取一个服务提供者发送回查事务状态请求，根据事务状态来决定是提交或回滚消息。

在RocketMQ中，消息在服务端的存储结构如下，每条消息都会有对应的索引信息，Consumer通过ConsumeQueue这个二级索引来读取消息实体内容，其流程如下：

RocketMQ源码分析13：事务消息

RocketMQ的具体实现策略是：写入的如果事务消息，对消息的Topic和Queue等属性进行替换，同时将原来的Topic和Queue信息存储到消息的属性中，正因为消息主题被替换，故消息并不会转发到该原主题的消息消费队列，消费者无法感知消息的存在，不会消费。其实改变消息主题是RocketMQ的常用“套路”，回想一下延时消息的实现机制。

  1. Commit和Rollback操作以及Op消息的引入 在完成一阶段写入一条对用户不可见的消息后，二阶段如果是Commit操作，则需要让消息对用户可见；如果是Rollback则需要撤销一阶段的消息。先说Rollback的情况。对于Rollback，本身一阶段的消息对用户是不可见的，其实不需要真正撤销消息（实际上RocketMQ也无法去真正的删除一条消息，因为是顺序写文件的）。但是区别于这条消息没有确定状态（Pending状态，事务悬而未决），需要一个操作来标识这条消息的最终状态。RocketMQ事务消息方案中引入了Op消息的概念，用Op消息标识事务消息已经确定的状态（Commit或者Rollback）。如果一条事务消息没有对应的Op消息，说明这个事务的状态还无法确定（可能是二阶段失败了）。引入Op消息后，事务消息无论是Commit或者Rollback都会记录一个Op操作。Commit相对于Rollback只是在写入Op消息前创建Half消息的索引。

  2. Op消息的存储和对应关系 RocketMQ将Op消息写入到全局一个特定的Topic中通过源码中的方法—TransactionalMessageUtil.buildOpTopic()；这个Topic是一个内部的Topic（像Half消息的Topic一样），不会被用户消费。Op消息的内容为对应的Half消息的存储的Offset，这样通过Op消息能索引到Half消息进行后续的回查操作.

RocketMQ源码分析13：事务消息

  1. Half消息的索引构建 在执行二阶段Commit操作时，需要构建出Half消息的索引。一阶段的Half消息由于是写到一个特殊的Topic，所以二阶段构建索引时需要读取出Half消息，并将Topic和Queue替换成真正的目标的Topic和Queue，之后通过一次普通消息的写入操作来生成一条对用户可见的消息。所以RocketMQ事务消息二阶段其实是利用了一阶段存储的消息的内容，在二阶段时恢复出一条完整的普通消息，然后走一遍消息写入流程。

  2. 如何处理二阶段失败的消息？ 如果在RocketMQ事务消息的二阶段过程中失败了，例如在做Commit操作时，出现网络问题导致Commit失败，那么需要通过一定的策略使这条消息最终被Commit。RocketMQ采用了一种补偿机制，称为“回查”。Broker端对未确定状态的消息发起回查，将消息发送到对应的Producer端（同一个Group的Producer），由Producer根据消息来检查本地事务的状态，进而执行Commit或者Rollback。Broker端通过对比Half消息和Op消息进行事务消息的回查并且推进CheckPoint（记录那些事务消息的状态是确定的）。

值得注意的是，rocketmq并不会无休止的的信息事务状态回查，默认回查15次，如果15次回查还是无法得知事务状态，rocketmq默认回滚该消息。

3. 从源码看事务消息

3.1 准备测试用例

/**
 * @author qiuguan
 */
public class TransactionProducer {

    public static void main(String[] args) throws MQClientException, InterruptedException {

        //TODO:定义事务消息的发送者，它继承了DefaultMQProducer（就是发送普通消息的类）
        TransactionMQProducer producer = new TransactionMQProducer(MQConstant.DEFAULT_PRODUCER_GROUP_NAME);
        //TODO:指定一个线程池，如果不指定MQ本身也会为我们创建一个，一般都会主动创建。消息回查时将用到
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client-transaction-msg-check-thread");
                return thread;
            }
        });

        producer.setNamesrvAddr(MQConstant.NAME_SERVER_ADDR);
        producer.setExecutorService(executorService);
        //TODO: 指定本地事务监听器
        TransactionListener transactionListener = new MyLocalTransactionCheckListener();
        producer.setTransactionListener(transactionListener);

        //TODO:producer启动
        producer.start();

        String[] tags = new String[]{"tagA", "tagB", "tagC"};

        for (int i = 0; i < 3; i++) {
            try {
                Message msg =
                        new Message(MQConstant.TX_TOPIC, tags[i], "KEY" + i,
                                ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));

                //第二个参数用于指定在执行本地事务时要使用的业务参数
                SendResult sendResult = producer.sendMessageInTransaction(msg, null);
                System.out.printf("发送事务消息结果%s%n", sendResult);

            } catch (MQClientException | UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }

        //先不要停掉生产者，观察事务回调和回查
        Thread.sleep(300000);

        producer.shutdown();
    }
}

参考官方用例

从用例中我们可以看到，消息的发送者从原来普通消息(批量消息,延迟消息)的发送者DefaultMQProducer变成了事务消息的专属发送者TransactionMQProducer

TransactionMQProducer 继承了 DefaultMQProducer

public class TransactionMQProducer extends DefaultMQProducer {
   //.....
}

其次，还需要指定一个本地事务监听器，当发送半消息成功时，我们使用 executeLocalTransaction 方法来执行本地事务，我们先看下代码示例：

/**
 * 本地事务监听器
 *
 * @author qiuguan
 */
public class MyLocalTransactionCheckListener implements TransactionListener {

    /**
     * @param msg
     * @param arg 他就是 producer.sendMessageInTransaction(msg, null); 的第二个参数null
     * @return
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        System.out.println("预提交消息成功：" + msg);

        /**
         * 假设接收到tagA的消息就表示扣款操作成功，
         * tagB的消息表示扣款失败，
         * tagC表示扣款结果不清楚，需要执行消息回查
         */
        if (StringUtils.equals("tagA", msg.getTags())) {
            return LocalTransactionState.COMMIT_MESSAGE;
        } else if (StringUtils.equals("tagB", msg.getTags())) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        } else if (StringUtils.equals("tagC", msg.getTags())) {
            return LocalTransactionState.UNKNOW;
        }
        return LocalTransactionState.UNKNOW;
    }


    /**
     * tagC 的本地事务状态是 UNKNOW, 所以会引起回查
     * 回查后是 COMMIT_MESSAGE, 那么则最终会提交该消息，被消费者看到消费
     * @param msg
     * @return
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        System.out.println("topic = " + msg.getTopic() + ", tag = " + msg.getTags() + " 消息回查");
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

这些我们先有个概念，后面在分析源码时，我们会看到它工作的时机。那么接下来我们就看下事务消息是如何发送的。

2.Producer启动

public class TransactionMQProducer extends DefaultMQProducer {

//TODO:.....略......

    @Override
    public void start() throws MQClientException {
        this.defaultMQProducerImpl.initTransactionEnv();
        super.start();
    }
}

事务消息相比其他消息多了一个初始化事务环境的方法。我们看下这个方法做了什么？

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

不难发现，它就是检查我们是否指定了线程池，如果没有指定，则MQ为我们指定一个。

然后是调用super.start()方法，既然是调用父类的，那就是我们前面熟悉的普通消息的启动过程，这里就不在赘述了。

3. 发送事务消息

public TransactionSendResult sendMessageInTransaction(final Message msg,
    final LocalTransactionExecuter localTransactionExecuter, final Object arg)
    throws MQClientException {
    //TODO:获取本地事务监听器，也就是前面demo中指定的那个
    TransactionListener transactionListener = getCheckListener();
    if (null == localTransactionExecuter && null == transactionListener) {
        throw new MQClientException("tranExecutor is null", null);
    }

    //TODO:事务消息不支持延迟
    if (msg.getDelayTimeLevel() != 0) {
        MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_DELAY_TIME_LEVEL);
    }

    Validators.checkMessage(msg, this.defaultMQProducer);

    SendResult sendResult = null;
    //TODO:设置事务消息的标识
    MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
    MessageAccessor.putProperty(msg, MessageConst.PROPERTY_PRODUCER_GROUP, this.defaultMQProducer.getProducerGroup());
    try {
        //TODO:同步发送消息，这个消息的发送过程和普通消息一模一样，就不赘述了
        //TODO:在broker端将会根据属性判断是否为事务消息，从而做特殊处理
        sendResult = this.send(msg);
    } catch (Exception e) {
        throw new MQClientException("send message Exception", e);
    }

    LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
    Throwable localException = null;
    //TODO:检查同步发送的消息是否成功
    switch (sendResult.getSendStatus()) {
        case SEND_OK: {
            try {
                //TODO:....省略部分判断........
                if (null != localTransactionExecuter) {
                    //TODO:这个API已经被废弃了
                    localTransactionState = localTransactionExecuter.executeLocalTransactionBranch(msg, arg);
                } else if (transactionListener != null) {
                    //TODO:使用这个API来执行本地事务，也就是我们指定的那个
                    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;
        //TODO:如果发送失败，则不会执行本地事务，直接设置为Rollback状态，broker将回滚
        case FLUSH_DISK_TIMEOUT:
        case FLUSH_SLAVE_TIMEOUT:
        case SLAVE_NOT_AVAILABLE:
            localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
            break;
        default:
            break;
    }

    try {
        //TODO:本地事务结束，设置事务状态，发给broker
        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;
}

从代码中可以看到，客户端发送事务消息主要做三件事：

  1. 设置事务消息属性，将消息(半事务消息)发往broker(这个发送过程就和普通消息一样).
  2. 如果消息发送成功，则执行本地事务(如果发送失败，则不会执行本地事务，直接回滚半事务消息).
  3. 本地事务执行完毕，则将本地事务执行的状态发送给broker,broker根据状态来判断是回滚还是提交.

我们通过下面这张简图可以看下： RocketMQ源码分析13：事务消息

接下来，我们再看下执行本地事务的过程，这个执行的就是前面我们给Producer设置的监听器并调用它重写的方法：

//TODO: 指定本地事务监听器
TransactionListener transactionListener = new MyLocalTransactionCheckListener();
producer.setTransactionListener(transactionListener);

那就看下本地事务的逻辑：

@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
    System.out.println("预提交消息成功：" + msg);

    /**
     * 假设接收到tagA的消息就表示扣款操作成功，
     * tagB的消息表示扣款失败，
     * tagC表示扣款结果不清楚，需要执行消息回查
     */
    if (StringUtils.equals("tagA", msg.getTags())) {
        return LocalTransactionState.COMMIT_MESSAGE;
    } else if (StringUtils.equals("tagB", msg.getTags())) {
        return LocalTransactionState.ROLLBACK_MESSAGE;
    } else if (StringUtils.equals("tagC", msg.getTags())) {
        return LocalTransactionState.UNKNOW;
    }
    return LocalTransactionState.UNKNOW;
}

接下来我们再看下事务结束的方法：endTransaction(....)

public void endTransaction(
    final Message msg,
    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());
    //TODO:最重要的就是这里：本地事务的执行结果状态是什么
    switch (localTransactionState) {
        case COMMIT_MESSAGE:
            requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_COMMIT_TYPE);
            break;
        case ROLLBACK_MESSAGE:
            requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_ROLLBACK_TYPE);
            break;
        //TODO:如果是UNKNOW 将会引起回查
        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;
    //TODO:将本地事务的执行结果状态发送给broker
    this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, requestHeader, remark,
        this.defaultMQProducer.getSendMsgTimeout());
}

继续跟进去看下：

public void endTransactionOneway(
    final String addr,
    final EndTransactionRequestHeader requestHeader,
    final String remark,
    final long timeoutMillis
) throws RemotingException, MQBrokerException, InterruptedException {
    //TODO：事务结束的code是：RequestCode.END_TRANSACTION
    RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.END_TRANSACTION, requestHeader);

    request.setRemark(remark);
    this.remotingClient.invokeOneway(addr, request, timeoutMillis);
}

事务消息相较于客户端来说，还是很简单的，也比较容易理解的，接下来我们就看下broker接收到半事务消息将做何处理？以及如何引起消息回查？

3.一阶段：Broker接收半事务消息

不要过分解读这个半事务消息，可以把它当做是一个普通消息，只不过broker在接收到这个普通消息时，由于它的属性中设置了事务的特性，所以会做些特殊处理，暂时不能被消息者消费，故叫半事务消息。

RocketMQ源码分析13：事务消息

前面我们说了，客户端在发送事务消息时，他和普通消息的发送流程是一模一样的，那么我们还是直接看我们熟悉的broker接收消息的处理器：SendMessageProcessor：

private CompletableFuture<RemotingCommand> asyncSendMessage(ChannelHandlerContext ctx, RemotingCommand request,
                                                            SendMessageContext mqtraceContext,
                                                            SendMessageRequestHeader requestHeader) {
                                                            
    //TODO:.....省略部分代码........                                                      
    
    //TODO:解析消息属性
    Map<String, String> origProps = MessageDecoder.string2messageProperties(requestHeader.getProperties());
    String transFlag = origProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);
    //TODO:判断是否有事务消息的属性
    if (transFlag != null && Boolean.parseBoolean(transFlag)) {

        //TODO: 事务消息
        putMessageResult = this.brokerController.getTransactionalMessageService().asyncPrepareMessage(msgInner);
    } else {

        //TODO: 普通消息
        putMessageResult = this.brokerController.getMessageStore().asyncPutMessage(msgInner);
    }
    return handlePutMessageResultFuture(putMessageResult, response, request, msgInner, responseHeader, mqtraceContext, ctx, queueIdInt);
}

普通消息我们已经很熟悉了，这里我们就直接看事务消息相关的流程：

public CompletableFuture<PutMessageResult> asyncPutHalfMessage(MessageExtBrokerInner messageInner) {
    return store.asyncPutMessage(parseHalfMessageInner(messageInner));
}
  1. 解析半事务消息 parseHalfMessageInner(.)
  2. 存储消息 store.asyncPutMessage(.)

那就先看下解析半事务消息的方法：

private MessageExtBrokerInner parseHalfMessageInner(MessageExtBrokerInner msgInner) {
    MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_TOPIC, msgInner.getTopic());
    MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_QUEUE_ID,
        String.valueOf(msgInner.getQueueId()));
    msgInner.setSysFlag(
        MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), MessageSysFlag.TRANSACTION_NOT_TYPE));
    msgInner.setTopic(TransactionalMessageUtil.buildHalfTopic());
    msgInner.setQueueId(0);
    msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
    return msgInner;
}

这里就看到了ROCKETMQ常用的手段，就是替换topic. 首先备份原始消息的topic到属性中，然后替换原始topic为框架内部的topic:RMQ_SYS_TRANS_HALF_TOPIC

然后将解析后的消息存储到commitlog中，这个存储过程和普通消息时一模一样的，这里也就不在继续赘述了。

到这里，一阶段的过程结束了，Producer成功发送了一条半事务消息到broker，由于broker替换了topic,所以消费者暂时无法消费这条消息。

4. 二阶段：Broker接收本地事务的状态

类比一阶段，二阶段就是这样子的： RocketMQ源码分析13：事务消息

前面我们有分析过，本地事务执行完毕后，将会产生一个 LocalTransactionState 状态，然后Producer将状态信息发给Broker,Broker接收到这个状态后，根据状态做相应的处理。LocalTransactionState 状态 有三个值：COMMIT_MESSAGE, ROLLBACK_MESSAGEM, UNKNOW(将会引起回查)，那么我们就根据这三种情况分别看下。

Broker端接收事务状态的处理器是：EndTransactionProcessor

public class EndTransactionProcessor extends AsyncNettyRequestProcessor implements NettyRequestProcessor {
        
        //TODO:......省略其他代码.......
   
        OperationResult result = new OperationResult();
        
        //TODO: 本地事务成功了
        if (MessageSysFlag.TRANSACTION_COMMIT_TYPE == requestHeader.getCommitOrRollback()) {
            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);
                    RemotingCommand sendResult = sendFinalMessage(msgInner);
                    if (sendResult.getCode() == ResponseCode.SUCCESS) {
                        this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage());
                    }
                    return sendResult;
                }
                return res;
            }
        } 
        //TODO: 本地事务回滚了
        else if (MessageSysFlag.TRANSACTION_ROLLBACK_TYPE == requestHeader.getCommitOrRollback()) {
            result = this.brokerController.getTransactionalMessageService().rollbackMessage(requestHeader);
            if (result.getResponseCode() == ResponseCode.SUCCESS) {
                RemotingCommand res = checkPrepareMessage(result.getPrepareMessage(), requestHeader);
                if (res.getCode() == ResponseCode.SUCCESS) {
                    this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage());
                }
                return res;
            }
        }
        response.setCode(result.getResponseCode());
        response.setRemark(result.getResponseRemark());
        return response;
    }
    
    //TODO:.....省略其他代码.........
    
}

4.1 本地事务提交：LocalTransactionState.COMMIT_MESSAGE

如果本地事务提交，则Broker的处理逻辑：

OperationResult result = new OperationResult();
if (MessageSysFlag.TRANSACTION_COMMIT_TYPE == requestHeader.getCommitOrRollback()) {
    //TODO:1.从commitlog中查询半事务消息
    result = this.brokerController.getTransactionalMessageService().commitMessage(requestHeader);
    if (result.getResponseCode() == ResponseCode.SUCCESS) {
        RemotingCommand res = checkPrepareMessage(result.getPrepareMessage(), requestHeader);
        if (res.getCode() == ResponseCode.SUCCESS) {
            //TODO:2.读取出Half消息，并将Topic和Queue替换成真正的目标的Topic和Queue
            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);
            //TODO:3.将替换后的消息再次写入commitlog中，此时消费者就可以消费了
            RemotingCommand sendResult = sendFinalMessage(msgInner);
            if (sendResult.getCode() == ResponseCode.SUCCESS) {
              //TODO:4.将半事务消息标记为删除
               this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage());
            }
            return sendResult;
        }
        return res;
    }
}

源码并不复杂，我们简单总结下：

  1. 从commitlog中读取出Half消息.
  2. 将Half消息中备份的原始topic和queueid取出，并替换掉原来的Half消息的topic和queueid.
  3. 将替换后的消息再次写入commitlog中，此时消费者就可以消费了.
  4. 将半事务消息标记为删除.

半事务消息标记为删除的具体做法是：新创建一个Message，topic是框架内部固定的RMQ_SYS_TRANS_OP_HALF_TOPIC，Tag 是 TransactionalMessageUtil.REMOVETAG，消息体的内容是半事务消息的consume queue offset, 然后将这条消息写入到commitlog中。

回查时，会用到这个Op消息

4.2 本地事务回滚：LocalTransactionState.ROLLBACK_MESSAGE

如果本地事务回滚，则Broker的处理逻辑：

OperationResult result = new OperationResult();
if (MessageSysFlag.TRANSACTION_COMMIT_TYPE == requestHeader.getCommitOrRollback()) {

    //TODO:....省略commit逻辑....
    
} else if (MessageSysFlag.TRANSACTION_ROLLBACK_TYPE == requestHeader.getCommitOrRollback()) 
    //TODO:1.从commitlog中查询半事务消息
    result = this.brokerController.getTransactionalMessageService().rollbackMessage(requestHeader);
    if (result.getResponseCode() == ResponseCode.SUCCESS) {
        RemotingCommand res = checkPrepareMessage(result.getPrepareMessage(), requestHeader);
        if (res.getCode() == ResponseCode.SUCCESS) {
          //TODO:将半事务消息标记为删除
           this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage());
        }
        return res;
    }
}

相比commit操作，rollback的逻辑更加简单，简单看下就行，这里就不在继续赘述了。

4.3 本地事务中间态：LocalTransactionState.UNKNOW

在 EndTransactionProcessor 处理器中，我们看到了 COMMIT 和 ROLLBACK, 但是并没有看到 UNKNOW 状态的处理逻辑，它实际上是通过异步线程TransactionalMessageCheckService 来完成处理的。

public class TransactionalMessageCheckService extends ServiceThread {
   
    //TODO:.....省略部分代码......

    @Override
    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!");
    }

    @Override
    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);
        //TODO:check操作
        this.brokerController.getTransactionalMessageService().check(timeout, checkMax, this.brokerController.getTransactionalMessageCheckListener());
        log.info("End to check prepare message, consumed time:{}", System.currentTimeMillis() - begin);
    }

}

继续跟进去看下：

public class TransactionalMessageServiceImpl implements TransactionalMessageService {
  
    //TODO:.....省略其他代码.........

    @Override
    public void check(long transactionTimeout, int transactionCheckMax,
        AbstractTransactionalMessageCheckListener listener) {
        try {
            String topic = TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC;
            Set<MessageQueue> msgQueues = transactionalMessageBridge.fetchMessageQueues(topic);
           
            //TODO......略.........

            for (MessageQueue messageQueue : msgQueues) {
                long startTime = System.currentTimeMillis();
                MessageQueue opQueue = getOpQueue(messageQueue);
                long halfOffset = transactionalMessageBridge.fetchConsumeOffset(messageQueue);
                long opOffset = transactionalMessageBridge.fetchConsumeOffset(opQueue);
                
                //TODO......略.........

                List<Long> doneOpOffset = new ArrayList<>();
                HashMap<Long, Long> removeMap = new HashMap<>();
                PullResult pullResult = fillOpRemoveMap(removeMap, opQueue, opOffset, halfOffset, doneOpOffset);
               
                //TODO......略.........

                int getMessageNullCount = 1;
                long newOffset = halfOffset;
                long i = halfOffset;
                while (true) {
                    if (System.currentTimeMillis() - startTime > MAX_PROCESS_TIME_LIMIT) {
                       
                         //TODO......略.........
                    }
                    if (removeMap.containsKey(i)) {
                         //TODO......略.........

                    } else {
                        GetResult getResult = getHalfMsg(messageQueue, i);
                        MessageExt msgExt = getResult.getMsg();
                        
                        //TODO......略.........

                       
                        List<MessageExt> opMsg = pullResult.getMsgFoundList();
                       
                        /**
                         * 什么样的消息需要回查？
                         * 1. 每一条半事务消息如果经过了commit/rollback, 那么对应就会创建一个Op消息，反过来说，如果没有Op消息，
                         *    但是有Half消息，那么说明是二阶段失败了（本地事务返回的状态码是UNKNOW），此时需要回查
                         *
                         * 2. 还有就是 TM在6秒内没有将最终确认状态发送给TC，此时也会触发回查
                         *
                         */
                        boolean isNeedCheck = (opMsg == null && valueOfCurrentMinusBorn > checkImmunityTime)
                            || (opMsg != null && (opMsg.get(opMsg.size() - 1).getBornTimestamp() - startTime > transactionTimeout))
                            || (valueOfCurrentMinusBorn <= -1);

                        if (isNeedCheck) {
                            //TODO......略.........

                            //TODO:...."消息回查"
                            listener.resolveHalfMsg(msgExt);

                        } else {
                            //TODO......略.........
                        }
                    }
                    newOffset = i + 1;
                    i++;
                }
                if (newOffset != halfOffset) {
                    transactionalMessageBridge.updateConsumeOffset(messageQueue, newOffset);
                }
                long newOpOffset = calculateOpOffset(doneOpOffset, opOffset);
                if (newOpOffset != opOffset) {
                    transactionalMessageBridge.updateConsumeOffset(opQueue, newOpOffset);
                }
            }
        } catch (Throwable e) {
            log.error("Check error", e);
        }

    }

 
    //TODO:.......省略其他代码...................
}

什么样的消息需要回查？

  1. 每一条半事务消息如果经过了commit/rollback, 那么对应就会创建一个Op消息，反过来说，如果没有Op消息，但是有Half消息，那么说明是二阶段失败了（本地事务返回的状态码是UNKNOW），此时需要回查.
  2. TM(也就是Producer)在6秒(默认6s)内没有将最终确认状态发送给TC(Broker)，此时也会触发回查.

那接下来我们就看下回查的整体过程:

4.3.1 Broker发起消息回查

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);
            }
        }
    });
}

继续跟进去：

public void sendCheckMessage(MessageExt msgExt) throws Exception {
    CheckTransactionStateRequestHeader checkTransactionStateRequestHeader = new CheckTransactionStateRequestHeader();
    
    //TODO:.....设置属性......
    
    Channel channel = brokerController.getProducerManager().getAvailableChannel(groupId);
    if (channel != null) {
       //TODO: Broker 给 Client 发送消息
       brokerController.getBroker2Client().checkProducerTransactionState(groupId, channel, checkTransactionStateRequestHeader, msgExt);
    } else {
        LOGGER.warn("Check transaction failed, channel is null. groupId={}", groupId);
    }
}

继续跟进去：

public void checkProducerTransactionState(
    final String group,
    final Channel channel,
    final CheckTransactionStateRequestHeader requestHeader,
    final MessageExt messageExt) throws Exception {
    RemotingCommand request =
    //TODO: code：RequestCode.CHECK_TRANSACTION_STATE
    RemotingCommand.createRequestCommand(RequestCode.CHECK_TRANSACTION_STATE, requestHeader);
    request.setBody(MessageDecoder.encode(messageExt, false));
    try {
        //TODO: 发送给生产者
        this.brokerController.getRemotingServer().invokeOneway(channel, request, 10);
    } catch (Exception e) {
        log.error("Check transaction failed because invoke producer exception. group={}, msgId={}, error={}",
                group, messageExt.getMsgId(), e.toString());
    }
}

到这里，Broker就将回查的请求发送给了Producer, 那么我们接下来就看下Producer 作何处理？

4.3.2 Producer 处理消息回查

Producer接收消息回查请求的是 ClientRemotingProcessor 类：

public class ClientRemotingProcessor extends AsyncNettyRequestProcessor implements NettyRequestProcessor {
    
    //TODO:......略.......
    
    @Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        switch (request.getCode()) {
            case RequestCode.CHECK_TRANSACTION_STATE:
                return this.checkTransactionState(ctx, request);
            
            //TODO:.....忽略其他case.......
            default:
                break;
        }
        return null;
    }
    
     //TODO:......略........
}

继续跟进去，来到 DefaultMQProducerImpl类的checkTransactionState(...)方法中，这方法内容比较多，但其实主要就做了3件事：

  1. 获取本地事务监听器(就是代码实例中指定的MyLocalTransactionCheckListener)
  2. 执行本地事务的回查
  3. 将回查结果上报给Broker
@Override
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();

        @Override
        public void run() {
            TransactionCheckListener transactionCheckListener = DefaultMQProducerImpl.this.checkListener();
            
            //TODO:1.获取事务监听器，就是demo中我们指定d的MyLocalTransactionCheckListener
            TransactionListener transactionListener = getCheckListener();
            if (transactionCheckListener != null || transactionListener != null) {
                LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
                
                //TODO: 2. 执行本地事务回查(省略其他if判断代码)
                localTransactionState = transactionListener.checkLocalTransaction(message);

                this.processTransactionState(
                    localTransactionState,
                    group,
                    exception);
            } else {
                log.warn("CheckTransactionState, pick transactionCheckListener by group[{}] failed", group);
            }
        }

        //TODO:这个方法和前面的endTransaction(....)方法的逻辑是一模一样的，就是将回查后的状态发给broker
        private void processTransactionState(
            final LocalTransactionState localTransactionState,
            final String producerGroup,
            final Throwable exception) {
            
            //TODO:.....省略诸多代码..........

            try {
               //TODO:3. 将回查结果上报给Broker
               DefaultMQProducerImpl.this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, thisHeader, remark,
                    3000);
            } catch (Exception e) {
                log.error("endTransactionOneway exception", e);
            }
        }
    };

    //TODO: 这个 checkExecutor 就是demo 示例中我们指定的线程池
    this.checkExecutor.submit(request);
}

其中回查的逻辑就在我们定义的事务监听器 MyLocalTransactionCheckListener 中：

/**
 * 本地事务监听器
 *
 * @author qiuguan
 */
public class LocalTransactionCheckListener implements TransactionListener {

    //TODO:.....省略本地事务逻辑......


    /**
     * tagC 的本地事务状态是 UNKNOW, 所以会引起回查
     * 回查后是 COMMIT_MESSAGE, 那么则最终会提交该消息，被消费者看到消费
     * @param msg
     * @return
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        System.out.println("topic = " + msg.getTopic() + ", tag = " + msg.getTags() + " 消息回查");
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

然后将回查后的状态，再次发送给Broker

DefaultMQProducerImpl.this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, thisHeader, remark,
                    3000);

到这里，其实整个事务消息的流程就结束了，我们可以通过一张简图看下： RocketMQ源码分析13：事务消息

5.事务消息的相关参数设置

  1. transactionTimeout=20，指定TM(Producer)在20秒内应将最终确认状态发送给TC(Broker)，否则引发消息回查。默认为6秒。
  2. transactionCheckMax=5，指定最多回查5次，超过后将丢弃消息并记录错误日志。默认15次。
  3. transactionCheckInterval=10，指定设置的多次消息回查的时间间隔为10秒。默认为60秒

其他参数还有一些，这里就不罗列了，在分析源码的过程中，我们也看到上述三个参数。

6.事务消息的使用总结

  1. 事务消息不支持延时消息和批量消息。
  2. 为了避免单个消息被检查太多次而导致半队列消息累积，MQ默认将单个消息的检查次数限制为 15 次，用户可以通过 transactionCheckMax 参数来修改此限制。如果已经回查某条消息超过 transactionCheckMax 次的话，则Broker将丢弃此消息，并打印错误日志。
  3. 事务消息将在参数 transactionTimeout 指定的时间长度之后被回查。当发送事务消息时，用户还可以通过设置用户属性 CHECK_IMMUNITY_TIME_IN_SECONDS 来改变这个限制，该参数优先于 transactionTimeout 参数。
  4. 事务消息的生产者 ID 不能与其他类型消息的生产者 ID 共享。与其他类型的消息不同，事务消息允许反向查询、MQ服务器能通过它们的生产者 ID 查询到消费者。

好了，关于RocketMQ的事务消息就记录到这里把。

限于作者水平，文中难免有错误之处，欢迎指正，勿喷，感谢感谢

转载自:https://juejin.cn/post/7137711358677614605
评论
请登录