RocketMQ事务消息实现分布式事务-编程知识

文章目录

- 简介
- 实现原理
- 实现逻辑

简介

RocketMQ事务消息
RocketMQ在4.3.0版中支持分布式事务消息，这里RocketMQ的事务消息是采用2PC(两段式协议) +补偿机制（消息回查）的分布式事务功能。提供消息发送与业务落库的一致性。
RocketMQ事务消息（Transactional Message）是指应用本地事务和发送消息操作可以被定义到全局事务中，要么同时成功，要么同时失败。RocketMQ的事务消息提供类似 X/Open XA 的分布式事务功能，通过事务消息能达到分布式事务的最终一致。

RocketMQ的事务消息实现方式主要包括以下几个步骤：
1 生产者发送half消息到Broker。half消息在消费者看来是不可见的，这样可以避免消费者消费到事务未提交的数据，类似于数据库的隔离级别读已提交级别，避免脏读。
2.生产者创建订单，根据创建订单成功与否，向Broker发送commit或rollback。
3.生产者还可以提供Broker回调接口，当Broker发现一段时间half消息没有收到任何操作命令，会主动调用此接口来查询订单是否创建成功。
4.一旦half消息commit了，消费者库存系统就会来消费，如果消费成功，则消息销毁，分布式事务成功结束。
5.如果消费失败，则根据重试策略进行重试，最后还失败则进入死信队列，等待进一步处理。

在这里插入图片描述

实现原理

RocketMQ的事务消息实现原理基于两阶段提交协议（Two-Phase Commit Protocol），具体流程如下：
1.发送准备消息：当一个事务消息需要发送时，生产者会发送一个准备消息，该消息包含了实际的业务数据和事务的标识符。
2.执行本地事务：接收到准备消息后，生产者会执行一个本地事务，如果本地事务执行成功，则返回COMMIT状态，否则返回ROLLBACK状态。
3.提交或回滚本地事务：当生产者返回COMMIT状态时，代表本地事务已经执行成功，此时，RocketMQ会将消息标记为可提交状态，并发送一个commit消息给消费者；当生产者返回ROLLBACK状态时，代表本地事务执行失败，此时，RocketMQ会将消息标记为可回滚状态，并发送一个rollback消息给消费者。
4. 消费者处理消息：消费者在接收到commit或rollback消息后，会根据消息的状态来执行相应的操作。
5.提交或回滚事务：如果所有的消费者都接收到了commit消息，则代表该事务消息已经提交，此时生产者会提交本地事务，否则，如果有任何一个消费者接收到了rollback消息，则代表该事务消息已经回滚，生产者会回滚本地事务。
事务消息的实现原理基于两阶段提交协议，这是一种经典的分布式事务协议，可以保证数据的一致性和可靠性。RocketMQ通过实现TransactionListener接口来支持事务消息，同时也提供了许多配置选项和工具类来方便用户进行使用和扩展。

实现逻辑

RocketMQ的事务消息通过TransactionListener接口来实现。下面是事务消息的基本实现步骤：
1.实现事务监听器：首先，你需要实现RocketMQ提供的TransactionListener接口，该接口包括两个方法：executeLocalTransaction和checkLocalTransaction。
○ executeLocalTransaction方法用于执行本地事务，当发送事务消息时，RocketMQ会调用此方法来执行本地事务。在该方法内部，你需要执行实际的业务逻辑，并根据执行结果返回事务状态，可以是提交、回滚或是未知状态。
○ checkLocalTransaction方法用于检查本地事务状态，当RocketMQ没有收到事务消息的确认或者取消时，会调用此方法来检查本地事务的状态，然后对消息进行处理。
2.发送事务消息：在发送事务消息时，你需要指定事务监听器，并在executeLocalTransaction方法中执行实际的业务逻辑，然后根据业务逻辑的执行结果返回事务状态。
3.事务状态检查：RocketMQ会定期调用checkLocalTransaction方法来检查本地事务的状态，然后对消息进行处理，例如提交或者回滚。
4.事务消息处理：在消息消费端，你需要根据消息的实际状态来执行相应的处理逻辑。
下面是一个简单的示例代码，演示了如何使用RocketMQ的事务消息：

public class TransactionProducer {public static void main(String[] args) throws MQClientException {DefaultMQProducer producer = new DefaultMQProducer("transaction_producer_group");producer.setNamesrvAddr("localhost:9876");// 设置事务监听器TransactionListener transactionListener = new MyTransactionListener();producer.setTransactionListener(transactionListener);// 启动生产者producer.start();try {// 创建事务消息Message msg = new Message("TopicTest", "TagA", "KEY1", "Hello, RocketMQ".getBytes());// 发送事务消息SendResult sendResult = producer.sendMessageInTransaction(msg, null);System.out.printf("%s%n", sendResult);} catch (MQClientException | UnsupportedEncodingException e) {e.printStackTrace();}}
}class MyTransactionListener implements TransactionListener {@Overridepublic LocalTransactionState executeLocalTransaction(Message msg, Object arg) {// 执行本地事务，例如数据库操作等// 返回本地事务的状态，可以是COMMIT、ROLLBACK或UNKNOW}@Overridepublic LocalTransactionState checkLocalTransaction(MessageExt msg) {// 检查本地事务状态，返回本地事务的状态，例如COMMIT、ROLLBACK或UNKNOW}
}