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事务消息内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

通过实例了解cookie机制特性及使用方法
这篇文章主要介绍了通过实例了解cookie机制特性及使用方法,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2020-09-09
一文带你彻底剖析Java中Synchronized原理
Synchronized是Java中的隐式锁，它的获取锁和释放锁都是隐式的，完全交由JVM帮助我们操作，在了解Synchronized关键字之前，首先要学习的知识点就是Java的对象结构，本文介绍的非常详细，需要的朋友可以参考下
2023-05-05
解决IDEA和CMD中java命令提示错误: 找不到或无法加载主类的问题
这篇文章主要介绍了解决IDEA和CMD中java命令提示错误: 找不到或无法加载主类的问题,本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2020-09-09
Mybatis配置之<typeAliases>别名配置元素解析
这篇文章主要介绍了Mybatis配置之<typeAliases>别名配置元素解析，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2021-07-07
SpringBoot整合Log4j2及配置步骤
这篇文章主要介绍了SpringBoot整合Log4j2以及配置详解,删除spring-boot-starter-parent默认使用spring-boot-starter-logging依赖，本文结合实例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2023-01-01
实例详解Java8函数式接口
本文给大家分析了Java8默认方法和函数式接口实例其它创建方式，需要的朋友跟着学习下吧。
2017-11-11
Spring Boot 如何生成微信小程序短连接及发送短信在短信中打开小程序操作
最近遇到这样的需求需要发送短信,通过短信中的短连接打开小程序操作,下面小编给大家分享Spring Boot 如何生成微信小程序短连接发送短信在短信中打开小程序操作,感兴趣的朋友跟随小编一起看看吧
2024-03-03
基于Ok+Rxjava实现断点续传下载
这篇文章主要为大家详细介绍了基于Ok+Rxjava实现断点续传下载，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2019-06-06
SpringBoot中事务失效的原因详解
这篇文章主要介绍了SpringBoot中事务失效的原因详解,spring中的事务是依赖AOP的,AOP是通过动态代理实现的,只有通过代理类访问的方法才能被拦截,需要的朋友可以参考下
2023-10-10
Java 数组分析及简单实例
这篇文章主要介绍了Java 数组分析及简单实例的相关资料,在Java中它就是对象，一个比较特殊的对象，需要的朋友可以参考下
2017-03-03