RocketMQ源码解析：事务消息是如何实现的？

用RocketMQ事务消息实现分布式事务

在分布式系统中为了保证数据的一致性，通常要使用分布式事务。分布式事务的解决方案有很多，比如TCC，SAGA，今天我们就来看一下如何用RocketMQ事务消息实现分布式事务？

RocketMQ实现分布式事务的流程如下

producer向mq server发送一个半消息
mq server将消息持久化成功后，向发送方确认消息已经发送成功，此时消息并不会被consumer消费
producer开始执行本地事务逻辑
producer根据本地事务执行结果向mq server发送二次确认，mq收到commit状态，将消息标记为可投递，consumer会消费该消息。mq收到rollback则删除半消息，consumer将不会消费该消息，如果收到unknow状态，mq会对消息发起回查
在断网或者应用重启等特殊情况下，步骤4提交的2次确认有可能没有到达mq server，经过固定时间后mq会对该消息发起回查
producer收到回查后，需要检查本地事务的执行状态
producer根据本地事务的最终状态，再次提交二次确认，mq仍按照步骤4对半消息进行操作

可以看到RocketMQ事务消息解决了Producer端事务执行和消息发送的一致性。并没有解决Producer端事务执行，消息发送和消息消费的一致性问题

当我们用RocketMQ实现分布式事务时，只需要实现TransactionListener接口即可，接口的2个方法作用如下

executeLocalTransaction，执行本地事务
checkLocalTransaction，回查本地事务状态

用官方的Demo演示一下具体的用法，写一个Producer类

public class TransactionProducer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        TransactionListener transactionListener = new TransactionListenerImpl();
        TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client-transaction-msg-check-thread");
                return thread;
            }
        });

        producer.setExecutorService(executorService);
        producer.setTransactionListener(transactionListener);
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.start();

        String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
        for (int i = 0; i < 10; i++) {
            try {
                Message msg =
                    new Message("TopicTest1234", tags[i % tags.length], "KEY" + i,
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.sendMessageInTransaction(msg, null);
                System.out.printf("%s%n", sendResult);

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

        for (int i = 0; i < 100000; i++) {
            Thread.sleep(1000);
        }
        producer.shutdown();
    }
}

写一个TransactionListener接口的实现类

public class TransactionListenerImpl implements TransactionListener {
    private AtomicInteger transactionIndex = new AtomicInteger(0);

    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        int value = transactionIndex.getAndIncrement();
        int status = value % 3;
        localTrans.put(msg.getTransactionId(), status);
        return LocalTransactionState.UNKNOW;
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        Integer status = localTrans.get(msg.getTransactionId());
        System.out.println("status " + status);
        if (null != status) {
            switch (status) {
                case 0:
                    return LocalTransactionState.UNKNOW;
                case 1:
                    return LocalTransactionState.COMMIT_MESSAGE;
                case 2:
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                default:
                    return LocalTransactionState.COMMIT_MESSAGE;
            }
        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

针对这个例子，所有的消息都会回查，因为executeLocalTransaction返回的都是UNKNOW，回查的时候status=1的数据会被消费，status=2的数据会被删除，status=0的数据会一直回查，直到超过默认的回查次数。

看到这，可能有人会问了，我们先执行本地事务，执行成功后再发送消息，不也可以实现类似的功能吗

其实这样做还是有可能会造成数据不一致的问题。假如本地事务执行成功，发送消息，由于网络延迟，消息发送成功，但是回复超时了，抛出异常，本地事务回滚。但是消息其实投递成功并被消费了，此时就会造成数据不一致的情况。

消息发送成功，但是提交事务失败了，例如事务提交超时，然后回滚，或者还没提交JVM宕机了，也会造成数据不一致

那消息投递到mq server，consumer消费失败怎么办？

如果是消费超时，重试即可。如果是由于代码等原因真的消费失败了，此时就得人工介入，重新手动发送消息，达到最终一致性。

源码解析

我将事务消息的执行流程画了如下一个执行流程图

Producer端处理流程

// DefaultMQProducerImpl#sendMessageInTransaction
public TransactionSendResult sendMessageInTransaction(final Message msg,
    final LocalTransactionExecuter localTransactionExecuter, final Object arg)
    throws MQClientException {
    TransactionListener transactionListener = getCheckListener();
    if (null == localTransactionExecuter && null == transactionListener) {
        throw new MQClientException("tranExecutor is null", null);
    }

    // ignore DelayTimeLevel parameter
    if (msg.getDelayTimeLevel() != 0) {
        MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_DELAY_TIME_LEVEL);
    }

    Validators.checkMessage(msg, this.defaultMQProducer);

    SendResult sendResult = null;
    // 添加属性，表明消息为 prepare 消息
    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);
    }

    LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
    Throwable localException = null;
    switch (sendResult.getSendStatus()) {
        case SEND_OK: {
            try {
                if (sendResult.getTransactionId() != null) {
                    msg.putUserProperty("__transactionId__", sendResult.getTransactionId());
                }
                String transactionId = msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
                if (null != transactionId && !"".equals(transactionId)) {
                    msg.setTransactionId(transactionId);
                }
                // 发送者成功发送PREPARED消息后，执行本地事务方法
                if (null != localTransactionExecuter) {
                    localTransactionState = localTransactionExecuter.executeLocalTransactionBranch(msg, arg);
                } else if (transactionListener != null) {
                    log.debug("Used new transaction 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;
        case FLUSH_DISK_TIMEOUT:
        case FLUSH_SLAVE_TIMEOUT:
        case SLAVE_NOT_AVAILABLE:
            localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
            break;
        default:
            break;
    }

    try {
        // 结束事务，提交或回滚，向rocketmq发送请求
        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;
}

事务消息的发送流程比较简单

首先给消息添加属性，表明这是一个半消息
发送消息，如果消息发送成功，则执行本地事务
根据本地事务的执行情况，向broker端发送结束事务的请求（commit rollback unknow）

Broker端处理流程

prepare消息存储

SendMessageProcessor#asyncSendMessage

当消息是一个半消息的时候，备份原来的topic和queueId到消息属性中，然后重新设置topic=RMQ_SYS_TRANS_HALF_TOPIC，queueId=0，存储到commitLog中，这样消费者就消费不了这条消息

接收结束事务请求

borker端通过EndTransactionProcessor来接收二阶段的消息，提交事务消息或者回滚事务消息

EndTransactionProcessor#processRequest

提交事务消息和回滚事务消息的逻辑差不多，先说提交事务消息的逻辑

根据消息偏移量从commitLog中找到消息（topic=RMQ_SYS_TRANS_HALF_TOPIC，queueId=0），然后将消息的topic和queueId重新设置为它原来的（存储的时候原来的topic和queueId已经备份到消息属性中了哈）
将消息重新存在commitLog中，这样消息就能被消费了哈
然后将消息存储在 topic 为 RMQ_SYS_TRANS_OP_HALF_TOPIC 的队列中，代表消息已经被处理（提交或回滚）

回滚事务失效的逻辑如下

根据消息偏移量从commitLog中找到消息（topic=RMQ_SYS_TRANS_HALF_TOPIC，queueId=0）
然后将消息存储在 topic 为 RMQ_SYS_TRANS_OP_HALF_TOPIC 的队列中，代表消息已经被处理（提交或回滚）

区别就是提交事务消息需要把原始的消息重新放到commitLog中，而回滚事务消息就不用

事务状态回查

TransactionalMessageCheckService#onWaitEnd

事务状态回查就是通过RMQ_SYS_TRANS_HALF_TOPIC（半消息存储的topic）和RMQ_SYS_TRANS_OP_HALF_TOPIC（已经处理的半消息的topic）来确定那些需要回查的半消息，根据回查结果确实是否要投递消息

参考博客

[1]https://juejin.cn/post/6844904193526857742#heading-7

本文标题：RocketMQ源码解析：事务消息是如何实现的？

本文链接：https://blog.quwenai.cn/post/8899.html

作者授权：除特别说明外，本文由 toobug 原创编译并授权 TOOBUG信息安全网刊载发布。

本文链接：https://blog.quwenai.cn/post/8899.html

分享到：

正文

RocketMQ源码解析：事务消息是如何实现的？

用RocketMQ事务消息实现分布式事务

源码解析

Producer端处理流程

Broker端处理流程

prepare消息存储

接收结束事务请求

事务状态回查

参考博客

相关阅读

【JVM】类加载器（二）：Tomcat 打破双亲委派

MYSQL自增主键丢失问题

【Java杂记】GUI技术：Swing基本使用及JFormDesigner插件安装使用

【Spring】日志：使用 spring 时日志的多种选择

发表评论取消回复

还没有评论，来说两句吧...

目录[+]