SpringBoot集成RocketMQ发送事务消息的原理解析

更新时间：2022年06月30日 15:48:40 作者：剑圣无痕

RocketMQ 的事务消息提供类似 X/Open XA 的分布事务功能，通过事务消息能达到分布式事务的最终一致，这篇文章主要介绍了SpringBoot集成RocketMQ发送事务消息,需要的朋友可以参考下

简介

RocketMQ 事务消息（Transactional Message）是指应用本地事务和发送消息操作可以被定义到全局事务中，要么同时成功，要么同时失败。RocketMQ 的事务消息提供类似 X/Open XA 的分布事务功能，通过事务消息能达到分布式事务的最终一致。

原理

RocketMQ事务消息通过异步确保方式，保证事务的最终一致性。设计的思想可以借鉴两个阶段提交事务。其执行流程图如下：

图片.png

发送方向MQ服务端发送消息。
MQ Server将消息持久化成功之后，向发送方 ACK 确认消息已经发送成功，此时消息为半消息。
发送方开始执行本地事务逻辑。
发送方根据本地事务执行结果向 MQ Server 提交二次确认(Commit 或是 Rollback)，MQ Server 收到 Commit 状态则将半消息标记为可投递，订阅方最终将收到该消息；MQ Server 收到 Rollback 状态则删除半消息，订阅方将不会接受该消息。
在断网或者是应用重启的特殊情况下，上述步骤4提交的二次确认最终未到达 MQ Server，经过固定时间后 MQ Server 将对该消息发起消息回查。
发送方收到消息回查后，需要检查对应消息的本地事务执行的最终结果。
发送方根据检查得到的本地事务的最终状态再次提交二次确认，MQ Server 仍按照步骤4对半消息进行操作。

具体实现

消费者

@Component
public class TransactionProduce
{
    private Logger logger = LoggerFactory.getLogger(getClass());
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    public void sendTransactionMessage(String msg)
    {
        logger.info("start sendTransMessage hashKey:{}",msg);
       
         Message message =new Message();
         message.setBody("this is tx message".getBytes());
         TransactionSendResult result=rocketMQTemplate.sendMessageInTransaction("test-tx-rocketmq", 
                 MessageBuilder.withPayload(message).build(), msg);
         
         //发送状态
         String sendStatus = result.getSendStatus().name();
         // 本地事务执行状态
         String localTxState = result.getLocalTransactionState().name();
         logger.info("send tx message sendStatus:{},localTXState:{}",sendStatus,localTxState);
    } 
}

说明：发送事务消息采用的是sendMessageInTransaction方法，返回结果为TransactionSendResult对象，该对象中包含了事务发送的状态、本地事务执行的状态等。

消费者

@Component
@RocketMQMessageListener(consumerGroup="test-txRocketmq-group",topic="test-tx-rocketmq", messageModel = MessageModel.CLUSTERING)
public class TransactionConsumer implements RocketMQListener<String>
{
    private Logger logger =LoggerFactory.getLogger(getClass());
    @Override
    public void onMessage(String message)
    {
        logger.info("send transaction mssage parma is:{}", message);
    }
}

说明：发送事务消息的消费者与普通的消费者一样没有太大的区别。

生产者消息监听器

发送事务消息除了生产者和消费者以外，我们还需要创建生产者的消息监听器，来监听本地事务执行的状态和检查本地事务状态。

@RocketMQTransactionListener
public class TransactionMsgListener implements RocketMQLocalTransactionListener
{
    private Logger logger = LoggerFactory.getLogger(getClass());
    /**
     * 执行本地事务
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg,
            Object obj)
    {
        logger.info("start invoke local rocketMQ transaction");
        RocketMQLocalTransactionState resultState = RocketMQLocalTransactionState.COMMIT;
        
        try
        {
            //处理业务
            String jsonStr = new String((byte[]) msg.getPayload(), StandardCharsets.UTF_8);
            logger.info("invoke msg content：{}",jsonStr);
        }
        catch (Exception e)
        {
            logger.error("invoke local mq trans error",e);
            resultState = RocketMQLocalTransactionState.UNKNOWN;
        }
        
        return resultState;
    }

    /**
     * 检查本地事务的状态
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg)
    {
        logger.info("start check Local rocketMQ transaction");
        
        RocketMQLocalTransactionState resultState = RocketMQLocalTransactionState.COMMIT;
        
        try
        {
            String jsonStr = new String((byte[]) msg.getPayload(), StandardCharsets.UTF_8);
            logger.info("check trans msg content：{}",jsonStr);
        }
        catch (Exception e)
        {
            resultState  = RocketMQLocalTransactionState.ROLLBACK;
        }
        return resultState;
    }
}

说明：RocketMQ本地事务状态由如下几种：

RocketMQLocalTransactionState.COMMIT：提交事务，允许消费者消费此消息。
RocketMQLocalTransactionState.ROLLBACK：回滚事务，消息将被删除，不允许被消费。
RocketMQLocalTransactionState.UNKNOWN：中间状态，代表需要进行检查来确定状态。

注意：Spring Boot2.0的版本之后，@RocketMQTransactionListener 已经没有了txProducerGroup属性，且sendMessageInTransaction方法也将其移除。所以在同一项目中只能有一个@RocketMQTransactionListener，不能出现多个，否则会报如下错误：

java.lang.IllegalStateException: rocketMQTemplate already exists RocketMQLocalTransactionListener

消息事务测试

正常测试

c.s.fw.mq.produce.TransactionProduce - product start sendTransMessage msg:{"userId":"zhangsann"}
c.s.f.m.p.TransactionMsgListener - start invoke local rocketMQ transaction
c.s.f.m.p.TransactionMsgListener - invoke local transaction msg content：{"topic":null,"flag":0,"properties":null,"body":"dGhpcyBpcyB0eCBtZXNzYWdl","transactionId":null,"keys":null,"tags":null,"delayTimeLevel":0,"waitStoreMsgOK":true,"buyerId":null}
c.s.fw.mq.produce.TransactionProduce - send tx message sendStatus:SEND_OK,localTXState:COMMIT_MESSAGE
c.s.f.m.consumer.TransactionConsumer - send transaction mssage parma is:{"topic":null,"flag":0,"properties":null,"body":"dGhpcyBpcyB0eCBtZXNzYWdl","transactionId":null,"keys":null,"tags":null,"delayTimeLevel":0,"waitStoreMsgOK":true,"buyerId":null}

说明：通过日志我们可以看出，执行的流程与上述的一致，执行成功后，消息执行成功返回的结果为SEND_OK，本地事务执行的状态为COMMIT_MESSAGE。

异常测试

如果在执行本地消息时出现异常，那么执行结果会是怎样？修改下本地事务执行的方法，让其出现异常。

代码调整

  @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg,
            Object obj)
    {
        logger.info("start invoke local rocketMQ transaction");
        RocketMQLocalTransactionState resultState = RocketMQLocalTransactionState.COMMIT;
        
        try
        {
            //处理业务
            String jsonStr = new String((byte[]) msg.getPayload(), StandardCharsets.UTF_8);
            logger.info("invoke local transaction msg content：{}",jsonStr);
             int c=1/0;
        }
        catch (Exception e)
        {
            logger.error("invoke local mq trans error",e);
            resultState = RocketMQLocalTransactionState.UNKNOWN;
        }
        
        return resultState;
    }

执行结果

c.s.fw.mq.produce.TransactionProduce - send tx message sendStatus:SEND_OK,localTXState:UNKNOW

从执行的结果可以看出，消息执行成功返回的结果为SEND_OK，本地事务执行的状态为:UNKNOW.所以消费端无法消费此消息。

总结

到此这篇关于SpringBoot集成RocketMQ发送事务消息的文章就介绍到这了,更多相关SpringBoot集成RocketMQ事务消息内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

mybatis中的if test判断入参的值问题
这篇文章主要介绍了mybatis中的if test判断入参的值问题，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2022-06-06
SpringCloud如何使用Eureka实现服务之间的传递数据
这篇文章主要介绍了SpringCloud使用Eureka实现服务之间的传递数据操作，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2021-06-06
Springboot hibernate envers使用过程详解
这篇文章主要介绍了Springboot hibernate envers使用过程详解,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2020-06-06
Java并发编程之ReentrantLock解析
这篇文章主要介绍了Java并发编程之ReentrantLock解析,ReentrantLock内容定义了一个抽象类Sync,继承自AQS,而不是自己去继承AQS,所有对ReentrantLock的操作都会转化为对Sync的操作,需要的朋友可以参考下
2023-12-12
maven解决包冲突方法详解
这篇文章主要介绍了maven解决包冲突方法详解，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2019-10-10
一文详解SpringBoot中CommandLineRunner接口
Spring Boot的CommandLineRunner接口是一个函数式接口,用于在Spring Boot应用程序启动后执行一些初始化操作,它提供了一个run方法,该方法在应用程序启动后被调用,本文给大家详细介绍了SpringBoot中CommandLineRunner接口,需要的朋友可以参考下
2023-10-10
MyBatis中获取Mysql数据库插入记录的主键值的实现
本文主要介绍了MyBatis中获取Mysql数据库插入记录的主键值的实现,包含了三种实现方式,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
2024-06-06
java ResourceBundle读取properties文件方式
这篇文章主要介绍了java ResourceBundle读取properties文件方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
2024-08-08
一篇带你解析入门LongAdder源码
LongAdder类是JDK1.8新增的一个原子性操作类。AtomicLong通过CAS算法提供了非阻塞的原子性操作,因为非常搞并发的请求下AtomicLong的性能是不能让人接受的
2021-06-06
Java spring MVC环境中实现WebSocket的示例代码
这篇文章主要介绍了Java spring MVC环境中实现WebSocket,本文通过示例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
2023-09-09