前言

Halo，我是白裤。

上一次我们学习了 RocketMQ 的文件读写的优化与设计，今天将带着大家一起了解 RocketMQ 的事务消息是怎么设计的，看看 RocketMQ 中事务消息与普通消息有什么区别，为什么使用 RocketMQ 事务消息可以达到数据最终一致性的效果。

分布式事务

在分布式开发中，通常我们会在一个接口中分别远程调用其他服务接口来完成业务操作，比如在订单支付完成后，需要更新用户积分、发优惠券等操作：

payOrder(){    //  更新用户积分（积分服务）  pointService.update();    //  发优惠券（营销服务）  promotionService.notify();  }

这就导致了分布式事务的产生，而分布式事务处理不好很容易出现数据的一致性问题，所以一般我们都尽量避开分布式事务，实在避开不了的，我们就采用业界比较成熟的一些解决方案来处理，比如 Seata，它支持 AT、TCC、SAGA、XA 等模式，而对于一些数据一致性没有那么强的要求的也就是非强一致性的事务我们通常可以采用最终一致性来处理，业界对于最终一致性的处理方案比较常用的有：本地消息表、事务消息等。

本地消息表方案指的是，创建一张消息日志表，将本地事务的业务操作和消息日志表放到一个事务中，这样可以达到事务原子性的条件，消息日志表和其他业务事务都是一起成功或者一起失败的，然后我们会再利用定时任务去扫描消息日志表，然后进行发送，此时消费方需要有幂等的处理，这样基本就达到了可靠消息一致性的结果。

事务消息方案指的是，利用事务消息的特性来完成原子性的操作，一般业界使用的是 RocketMQ 的事务消息，它通过提供消息发送与本地事务的原子性操作以及消息的持久化能力来达到可靠消息一致性的保证。

消息比较

普通消息

我们先来看看，当使用普通消息的时候，会有哪些问题。

第一种情况，先发送消息，再执行本地事务

payOrder(){
  //  发送订单支付成功的消息  producer.sendMsg();    //  更新订单信息  updateOrderInfo();    }

假如消息发送成功，但是订单信息更新失败，最终导致数据的不一致性。

第二种情况，先执行本地事务，再发送消息

payOrder(){    //  更新订单信息  updateOrderInfo();    //  发送订单支付成功的消息  producer.sendMsg();  }

假如订单信息更新成功，消息发送超时的时候，发生本地事务回滚，但是此时有可能消息其实已经发送成功，最终导致了数据的不一致。

可以看到，使用普通消息并不能解决本地事务与消息的原子性问题，不能一起成功或者一起失败，最终都会导致数据的不一致。

事务消息

接下来我们来看看用事务消息去处理。

payOrder(){  //  发送事务半消息  producer.sendHalfMsg();    //  更新订单信息  updateOrderInfo();    //  确认事务消息的提交/回滚  producer.commit()/rollback();  }

先发送事务半消息，这里类似二阶段的预提交，然后更新本地事务，如果本地事务执行失败，则进行消息的回滚，如果本地事务执行成功，则进行消息的提交，解决了消息不可回滚的问题，同时也保证了消息发送与本地事务的原子性和数据的一致性。

RocketMQ 事务消息使用

那么事务消息如何使用呢？下面是 RocketMQ 事务消息的使用案例：

实现监听器

首先实现一个监听器，实现 RocketMQ 的 TransactionListener 接口，然后实现两个方法，分别是 executeLocalTransaction(Message msg, Object arg)方法和 checkLocalTransaction(MessageExt msg)方法。

• executeLocalTransaction(Message msg, Object arg)方法：执行你的本地事务，然后根据情况返回一个事务状态（提交/回滚）。

• checkLocalTransaction(MessageExt msg)方法：RocketMQ 的回查方法，当你的消息预提交之后，如果 RocketMQ 一直没有收到你的确认请求（提交/回滚），那么 RocketMQ 会去调用此方法回查并确认提交状态。

public class PayOrderSuccessListenerImpl implements TransactionListener {
    @Override    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {        try {            //  更新订单信息            updateOrderInfo();            return LocalTransactionState.COMMIT_MESSAGE;        } catch (Exception e) {            log.error("PayOrderSuccessMsg execute local transaction error", e);            return LocalTransactionState.ROLLBACK_MESSAGE;        }    }
    @Override    public LocalTransactionState checkLocalTransaction(MessageExt msg) {        //  查询订单支付状态        Integer status = obtainOrderPayStatus();        if (Objects.nonNull(status)) {            switch (status) {                case 0:                    return LocalTransactionState.ROLLBACK_MESSAGE;                case 1:                    return LocalTransactionState.COMMIT_MESSAGE;                default:                    return LocalTransactionState.UNKNOW;            }        }        return LocalTransactionState.UNKNOW;
    }}

LocalTransactionState 分别有三种状态：COMMIT_MESSAGE（提交消息）、ROLLBACK_MESSAGE（回滚消息）、UNKNOW（未知）

构建消息及发送

创建事务消息生产者，这里注意跟普通消息不一样的是，构建的生产者是 TransactionMQProducer，而非普通的 DefaultMQProducer，然后将监听器设置到生产者中，执行消息的发送即可。

public class TransactionProducerTest {    public static void main(String[] args) throws MQClientException, InterruptedException {        //  监听器实现        TransactionListener transactionListener = new PayOrderSuccessListenerImpl();        //  事务消息生产者        TransactionMQProducer producer = new TransactionMQProducer("trans_test");        //  设置监听器        producer.setTransactionListener(transactionListener);        //  启动        producer.start();
        try {            //  构建消息体            Message msg =                    new Message("TopicTest", "TagTest", "KEY",                            ("Hello RocketMQ").getBytes(RemotingHelper.DEFAULT_CHARSET));            //  发送事务消息            SendResult sendResult = producer.sendMessageInTransaction(msg, null);            System.out.printf("%s%n", sendResult);
        } catch (MQClientException | UnsupportedEncodingException e) {            e.printStackTrace();        }
        //  关闭        producer.shutdown();    }}

可以看到，RocketMQ 事务消息的使用非常的方便，可以达到分布式事务的最终一致性。

RocketMQ 事务消息流程

我们来看下 RocketMQ 的事务消息流程图：

如上图所示：

【1】生产者（订单服务）将事务半消息发送至 RocketMQ Broker。

【2】RocketMQ Broker 将消息持久化成功之后，向生产者（订单服务）返回 ack 确认半消息发送成功。

【3】生产者（订单服务）执行本地事务逻辑（更新订单信息）。

【4】生产者（订单服务）向RocketMQ Broker确认结果（提交/回滚），RocketMQ Broker收到确认结果后进行以下处理：

​ ①如果结果为 Commit：走【8】步骤，RocketMQ Broker将半事务消息标记为可投递，并投递给消费者。

​ ②如果结果为 Rollback：走【9】步骤，RocketMQ Broker将回滚事务，不会将半事务消息投递给消费者。

【5】如果RocketMQ Broker一直未收到生产者（订单服务）的结果确认，或者收到的确认结果是 Unknown 状态的话，将会进行消息确认结果的回查。

【6、7】此时生产者（订单服务）会进行本地事务的状态（订单支付状态）检查，然后根据本地事务状态（订单支付状态）再次发起事务消息的确认结果。

​ ①如果结果为 Commit：走【8】步骤，RocketMQ Broker将半事务消息标记为可投递，并投递给消费者。

​ ②如果结果为 Rollback：走【9】步骤，RocketMQ Broker将回滚事务，不会将半事务消息投递给消费者。

​ ③如果结果为 Unknown：继续等待下一次的状态反查。

RocketMQ 事务消息原理

那么 RocketMQ 的事务消息是怎么实现的呢？为什么半消息会不被消费者所感知消费？

其实当生产者向 RocketMQ 发送事务半消息的时候，RocketMQ 会将这个半消息投递到一个 Topic 为RMQ_SYS_TRANS_HALF_TOPIC的队列中，由于消费者们对RMQ_SYS_TRANS_HALF_TOPIC队列没有进行订阅，所以是不会消费到此半消息的，RocketMQ 内部会启动定时任务去消费扫描RMQ_SYS_TRANS_HALF_TOPIC这个队列，如果 RocketMQ 在此消息事务超时之前还未接收到提交或者回滚确认，则会进行反查处理，再根据反查确认结果进行对应的处理，如果是 Commit 状态，才将此半消息真正投递到消息发送时定义的 Topic 的队列中，对应的订阅者即消费者们才得以消费到此消息，如果是 Rollback 状态，则会回滚事务消息，而如果状态一直是 Unknown 状态的话，RocketMQ 反查一定次数（默认 15 次）后将放弃该事务消息，直接回滚事务消息。

小结

好了，今天 RocketMQ 的事务消息的设计就学习到这里，接下来我们做个小结吧。

首先我们先简单了解了分布式事务的常见问题，然后了解到解决分布式事务的一些方案，而我们学习的 RocketMQ 事务消息是解决分布式事务的其中一种方式，接着我们了解了普通消息和事务消息的区别，然后分别学习了 RocketMQ 事务消息使用、消息流程、消息原理，到这里，我们对于 RocketMQ 事务消息的了解就更进一步了，以后碰到类似的使用与机制理解我们就更加熟悉啦。