Java小案例-RocketMQ的11种消息类型，你知道几种？（事务消息）

news2024/12/24 9:02:59

前言

上一节给大家讲了Rocket的延迟消息，这一节和大家聊一下事务消息，关于延迟消息大家可以点下面这个链接直接看。

事务消息

事务消息是RocketMQ提供的一种类似X/Open XA的分布式事务功能。通过RocketMQ的事务消息，可以达到分布式事务的最终一致。

事务消息用起来也比较简单，如下所示：

public class TransactionMessageDemo {

    public static void main(String[] args) throws Exception {
        TransactionMQProducer transactionMQProducer = new TransactionMQProducer("sanyouProducer");
        transactionMQProducer.setNamesrvAddr("192.168.200.143:9876");

        //设置事务监听器
        transactionMQProducer.setTransactionListener(new TransactionListener() {

            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                //处理本次事务
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                //检查本地事务
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        transactionMQProducer.start();

        Message message = new Message("sanyouTopic", "java日记".getBytes());

        //发送消息
        transactionMQProducer.sendMessageInTransaction(message, new Object());
    }

}

事务消息发送相对于前面的例子主要有以下不同：

将前面的DefaultMQProducer换成TransactionMQProducer
需要设置事务的监听器TransactionListener，来执行本地事务
发送方法改成 sendMessageInTransaction

为什么要这么改，接下来我们来讲讲背后的实现原理

上一节在说延迟消息的时候提到，RocketMQ使用到了SCHEDULE_TOPIC_XXXX这个中转Topic，来偷梁换柱实现延迟消息

不仅仅是延迟消息，事务消息其实也是这么干的，它也会进行偷梁换柱，将消息先存在RMQ_SYS_TRANS_HALF_TOPIC这个Topic下，同时也会将消息真正的Topic和队列id存到额外信息中，操作都是一样滴

由于消息不在真正目标的Topic下，所以这条消息消费者也是消费不到滴

当消息成功存储之后，服务端会向生产者响应，告诉生产者我消息存储成功了，你可以执行本地事务了

之后生产者就会执行本地执行事务，也就是执行如下方法

TransactionListener#executeLocalTransaction

当本地事务执行完之后，会将执行的结果发送给服务端

服务端会根据事务的执行状态来执行对应的处理结果

commit：提交事务消息，跟延迟消息一样，重新构建一条消息，Topic和队列id都设置成消息真正的Topic和队列id，然后重新存到CommitLog文件，这样消费者就可以消费到消息了
rollback：回滚消息，其实并没有实际的操作，因为消息本身就不在真正的Topic下，所以消费者压根就消费不到，什么都不做就可以了
unknown：本地事务执行异常时就是这个状态，这个状态下会干一些事，咱们后面再说

所以在正常情况下，事务消息整个运行流程如下图所示