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

您可能感兴趣的文章:

idea使用war以及war exploded的区别说明
本文详细解析了war与warexploded两种部署方式的差异及步骤,war方式是先打包成war包,再部署到服务器上；warexploded方式是直接把文件夹、class文件等移到Tomcat上部署,支持热部署,开发时常用,文章分别列出了warexploded模式和war包形式的具体操作步骤
2024-10-10
Java数据结构及算法实例：三角数字
这篇文章主要介绍了Java数据结构及算法实例：三角数字,本文直接给出实现代码,代码中包含详细注释,需要的朋友可以参考下
2015-06-06
java正则匹配读取txt文件提取特定开头和结尾的字符串
通常我们可以直接通过文件流来读取txt文件的内容,但有时候也会遇到问题,下面这篇文章主要给大家介绍了关于java正则匹配读取txt文件提取特定开头和结尾的字符串的相关资料,需要的朋友可以参考下
2022-11-11
Java仿淘宝首页分类列表功能的示例代码
这篇文章主要介绍了仿淘宝分类管理功能的示例代码，具有很好的参考价值，希望对大家有所帮助，也给大家做个参考
2018-05-05
Android应用开发之将SQLite和APK一起打包的方法
这篇文章主要介绍了Android应用开发之将SQLite和APK一起打包的方法,文章时间较早,尽管现在开发环境已大都迁移至Android Studio上,但打包原理依然相同,需要的朋友可以参考下
2015-08-08
Java中关于Map四种取值方式
这篇文章主要介绍了Java中关于Map四种取值方式，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2023-03-03
Maven配置多仓库无效的解决
在项目中使用Maven管理jar包依赖往往会出现很多问题，所以这时候就需要配置Maven多仓库，本文介绍了如何配置以及问题的解决
2021-05-05
java 中modCount 详解及源码分析
这篇文章主要介绍了java 中modCount 详解及源码分析的相关资料,需要的朋友可以参考下
2017-02-02
深入浅析hbase的优点
本文讲述了HBase的特征和它的优点，并简要回顾了行键设计的重点之处，它还向你展示了如何在本地配置HBase环境，使用命令创建表、插入数据、检索指定行以及最后如何进行scan操作，感兴趣的朋友一起看看吧
2017-09-09
Java中的反射机制示例详解
反射就是把Java类中的各个成分映射成一个个的Java对象。本文将通过示例详细讲解Java中的反射机制，感兴趣的小伙伴可以跟随小编学习一下
2022-03-03