关键词：
RocketMQ提供了事务消息，通过事务消息就能达到分布式事务的最终一致，从而实现了可靠消息服务。
一、事务消息的实现步骤
事务消息发送步骤：
1. 发送方将半事务消息发送至RocketMQ服务端。
2. RocketMQ服务端将消息持久化之后，向发送方返回Ack确认消息已经发送成功。由于消息为半事务消息，在未收到生产者对该消息的二次确认前，此消息被标记成“暂不能投递”状态。
3. 发送方开始执行本地事务逻辑。
4. 发送方根据本地事务执行结果向服务端提交二次确认（Commit 或是 Rollback），服务端收到Commit 状态则将半事务消息标记为可投递，订阅方最终将收到该消息；服务端收到 Rollback 状态则删除半事务消息，订阅方将不会接受该消息。
事务消息回查步骤：
1. 在断网或者是应用重启的特殊情况下，上述步骤4提交的二次确认最终未到达服务端，经过固定时间后服务端将对该消息发起消息回查。
2. 发送方收到消息回查后，需要检查对应消息的本地事务执行的最终结果。
3. 发送方根据检查得到的本地事务的最终状态再次提交二次确认，服务端仍按照步骤4对半事务消息进行操作。
二、程序实现
事务消息处理类需要继承RocketMQLocalTransactionListener类。该类的executeLocalTransaction方法负责在接到RocketMQ服务端的Ack确认消息后执行本地方法，也就是事务消息发送步骤中的步骤3。该类的checkLocalTransaction方法负责，在断网或者是应用重启的特殊情况下，执行RocketMQ服务端的消息回查，也就是事务消息回查步骤中的步骤2。
此外，要使该类生效，还需要加@RocketMQTransactionListener注解。这里有个要特别注意的地方。在2.1.0版本前，这个注解有一个属性txProducerGroup，可以用多个@RocketMQTransactionListener来监听不同的txProducerGroup来发送不同类型的事务消息到topic。但是现在在一个项目中，如果你在一个project中写了多个@RocketMQTransactionListener，项目将不能启动，启动会报错。产生这个问题的原因据说是，当使用RocketMQTemplate并发的执行事务时，非常容易出现"illegal state"的异常，原因是一个TransactionProducer在执行事务时不能被共享。所以，必须使用同一个TransactionMQProducer来发送所有类型的事务消息。当然同理也就必须使用一个侦听器处理所有的消息了。
既然必须使用同一个TransactionMQProducer，对于比较大的应用，业务场景很多，就会造成混乱。这里我给出一个方案抛砖引玉。TransactionMQProducer在发送消息时，是可以传递参数对象和指定消息头的。可以把要执行的本地方法的bean名和方法名放进去。
//发送半事务消息
TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction(
topicAndTag,
MessageBuilder.withPayload(msg)
.setHeader(Constants.TX_ID_HEADER_NAME, msg.getTxId())
.setHeader(Constants.CHECK_BEAN_ID_HEADER_NAME, def.getCheckBeanId())
.setHeader(Constants.BIZ_ID_HEADER_NAME, msg.getBizId())
.build(),
def
);其中def就是参数对象，可以自定义对象，这里是我自定义的TransactionMsgDefinationDto类，可以把想传递的信息放进去，最重要的是要执行的本地方法的bean名和方法名和方法执行参数：executeBeanId（bean名）、executeBeanMethod（方法名）、executeBeanParams（方法执行参数）。该对象可以传给RocketMQLocalTransactionListener的executeLocalTransaction方法，然后通过反射执行。
@Override
public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg)
try
//保存消息记录
String body = new String((byte[]) msg.getPayload(), StandardCharsets.UTF_8);
JSONObject jsonBody = JSONObject.parseObject(body);
BaseMsgDto dto = JSONObject.toJavaObject(jsonBody, BaseMsgDto.class);//(BaseMsgDto)msg.getPayload();
TransactionMsgDefinationDto def = (TransactionMsgDefinationDto)arg;
ProducerLog producerLog = BeanCopyUtils.copyProperties(def, ProducerLog::new);
String[] tags = def.getMsgTags();
if(tags !=null && tags.length > 0)
StringBuilder tag = new StringBuilder();
for(int i = 0; i<tags.length; i++)
tag.append(tags[0]);
if(i != tags.length-1)
tag.append("||");
producerLog.setMsgTag(tag.toString());
producerLog.setBizId(dto.getBizId());
producerLog.setTxId(dto.getTxId());
producerLog.setBizType(dto.getBizType());
producerLog.setGroupName(dto.getProducerGroup());
producerLog.setMsgBody(body);
producerLogService.save(producerLog);
//执行事务方法
SpringUtil.invokeBeanMethod(def.getExecuteBeanId(), def.getExecuteBeanMethod(), def.getExecuteBeanParams());
return RocketMQLocalTransactionState.COMMIT;
catch (Exception e)
logger.error("发生错误：", e);
return RocketMQLocalTransactionState.UNKNOWN;
放在消息头header中的数据可以传递给RocketMQLocalTransactionListener的checkLocalTransaction方法，然后同样通过反射执行。
@Override
public RocketMQLocalTransactionState checkLocalTransaction(Message msg)
try
String txId = (String)msg.getHeaders().get(Constants.TX_ID_HEADER_NAME);
String checkBeanId = (String)msg.getHeaders().get(Constants.CHECK_BEAN_ID_HEADER_NAME);
Long bizId = Long.parseLong((String)msg.getHeaders().get(Constants.BIZ_ID_HEADER_NAME));
//执行检查方法
Boolean ret = (Boolean)SpringUtil.invokeBeanMethod(checkBeanId, "check", new Object[]bizId, txId);
if(ret.booleanValue())
return RocketMQLocalTransactionState.COMMIT;
else
return RocketMQLocalTransactionState.ROLLBACK;
catch (Exception e)
logger.error("发生错误：", e);
return RocketMQLocalTransactionState.UNKNOWN;
rocketmq（09）——发送事务消息(代码片段)
发送事务消息RocketMQ支持发送事务消息，它的事务消息是基于二阶段提交机制实现的。当发送的消息是事务消息时，只有对应的消息被提交了才能被消费者进行消费。发送事务消息时生产者需要使用TransactionMQProducer，... 查看详情
rocketmq事务消息机制(代码片段)
RocketMQ提供了事务消息，通过事务消息就能达到分布式事务的最终一致，从而实现了可靠消息服务。一、事务消息的实现步骤事务消息发送步骤：1.发送方将半事务消息发送至RocketMQ服务端。2.RocketMQ服务端将消息持久... 查看详情
rocketmq支持事务消息机制(代码片段)
事务消费我们经常支付宝转账余额宝，这是日常生活的一件普通小事，但是我们思考支付宝扣除转账的钱之后，如果系统挂掉怎么办，这时余额宝账户并没有增加相应的金额，数据就会出现不一致状况了。上... 查看详情
rocketmq实现事务消息方案(代码片段)
RocketMQ是一个来自阿里巴巴的分布式消息中间件，于2012年开源，并在2017年正式成为Apache顶级项目。据了解，包括阿里云上的消息产品以及收购的子公司在内，阿里集团的消息产品全线都运行在RocketMQ之上，并且最近几年的双十一... 查看详情
rocketmq源码分析之rocketmq事务消息实现原理中篇----事务消息状态回查(代码片段)
上节已经梳理了RocketMQ发送事务消息的流程（基于二阶段提交），本节将继续深入学习事务状态消息回查，我们知道，第一次提交到消息服务器时消息的主题被替换为RMQ_SYS_TRANS_HALF_TOPIC，本地事务执行完后如果返回本地事务状态... 查看详情
rocketmq源码分析之rocketmq事务消息实现原理中篇----事务消息状态回查(代码片段)
上节已经梳理了RocketMQ发送事务消息的流程（基于二阶段提交），本节将继续深入学习事务状态消息回查，我们知道，第一次提交到消息服务器时消息的主题被替换为RMQ_SYS_TRANS_HALF_TOPIC，本地事务执行完后如果返回本地事务状态... 查看详情
rocketmq事务消息入门介绍(代码片段)
说明周五的时候发了篇：Rocketmq4.3支持事务啦！！！，趁着周末的时候把相关内容看了下，下面的主要内容就是关于RocketMQ事务相关内容介绍了。说明：今天这篇仅仅是入门介绍，并没有涉及到很多细节，先把大概流程说明白，... 查看详情
rocketmq事务消息篇之事务消息的使用(代码片段)
前言在RocketMQ事务消息篇（一）之事务消息的介绍里对RocketMQ的事务消息作了相关说明，本文提供一些基本的开发示例。java示例依赖<dependencies><dependency><groupId>org.apache.rocketmq</groupId><artifactId>ro 查看详情
rocketmq事务消息篇之事务消息源码分析(代码片段)
前言RocketMQ事务消息篇（一）之事务消息的介绍RocketMQ事务消息篇（二）之事务消息的使用本文继前两篇对事务消息源码进行分析。事务消息处理基本流程在介绍事务消息的时候，画了一个简单的流程图说明事... 查看详情
rocketmq事务消息原理(代码片段)
一、RocketMQ事务消息原理： RocketMQ在4.3版本之后实现了完整的事务消息，基于MQ的分布式事务方案，本质上是对本地消息表的一个封装，整体流程与本地消息表一致，唯一不同的就是将本地消息表存... 查看详情
30rocketmq事务消息的代码实现细节(代码片段)
...。1.发送half事务消息出去packagecom.mqTrsMessage;importorg.apache.rocketmq.client.producer.DefaultMQProducer;importorg.apache.rocketmq.clien 查看详情
11springboot整合rocketmq实现事务消息(代码片段)
事务消息是RocketMQ提供的非常重要的一个特性，在4.x版本之后开源，可以利用事务消息轻松地实现分布式事务。RocketMQ在其消息定义的基础上，对事务消息扩展了两个相关的概念：Half(Prepare)Message——半消息(预处理... 查看详情
rocketmq使用事务消息(代码片段)
目录说明原理事务消息处理流程生产端消费端说明事务消息：1、不支持延时消息和批量消息2、如果消息没有及时提交，默认check15次，可以通过Broker的transactionCheckMax参数配置次数。如果超时15次依然没有得到明确结果... 查看详情
rocketmq知识图谱总结(代码片段)
文章目录前言RocketMQ组成提高性能地方过期机制高可用NameSpace刷盘消息储存消息发送消息消费事务消息前言本博客是观看中通架构师丁威老师《rocketmq技术内幕》总结，以及网上一些技术博客，仅用于个人学习以及分享个... 查看详情
rocketmq事务消息实战(代码片段)
RocketMQ事务消息阅读目录指引：RocketMQ源码分析之从官方示例窥探RocketMQ事务消息实现基本思想RocketMQ源码分析之RocketMQ事务消息实现原理上篇RocketMQ源码分析之RocketMQ事务消息实现原理中篇----事务消息状态回查RocketMQ源码分析... 查看详情
rocketmq源码分析之从官方示例窥探：rocketmq事务消息实现基本思想(代码片段)
RocketMQ4.3.0版本开始支持事务消息，后续分享将开始将剖析事务消息的实现原理。首先从官方给出的Demo实例入手，以此通往RocketMQ事务消息的世界中。官方版本未发布之前，从apacherocketmq第一个版本上线后，代码中存在与事务消息... 查看详情
rocketmq事务消息详解(代码片段)
...#x1f34a;Java学习：Java从入门到精通总结🍊深入浅出RocketMQ设计思想：深入浅出RocketMQ设计思想🍊绝对不一样的职场干货：大厂最佳实践经验指南📆最近更新：2023年4月9日🍊个人简介：通信工程本... 查看详情
七.rocketmq极简入门-rocketmq事务消息(代码片段)
概述如果业务只涉及到一个数据库的写操作，我们只需要保证这一个事物的提交和回滚，这种事务管理叫传统事物或本地事务，如果业务涉及到多个数据库(多个服务)的写操作，我们需要保证多个数据库同时提交... 查看详情