关键词：
RocketMQ事务消息阅读目录指引：
RocketMQ源码分析之从官方示例窥探RocketMQ事务消息实现基本思想
RocketMQ源码分析之RocketMQ事务消息实现原理上篇
RocketMQ源码分析之RocketMQ事务消息实现原理中篇----事务消息状态回查
RocketMQ源码分析之事务消息实现原理下篇-消息服务器Broker提交回滚事务实现原理
RocketMQ事务消息实战
我们以一个订单流转流程来举例，例如订单子系统创建订单，需要将订单数据下发到其他子系统（与第三方系统对接）这个场景，我们通常会将两个系统进行解耦，不直接使用服务调用的方式进行交互。其业务实现步骤通常为：
1、A系统创建订单并入库。
2、发送消息到MQ。
3、MQ消费者消费消息，发送远程RPC服务调用，完成订单数据的同步。
1、方案一
方案弊端：
1、如果消息发送成功，在提交事务的时候JVM突然挂掉，事务没有成功提交，导致两个系统之间数据不一致。
2、由于消息是在事务提交之前提交，发送的消息内容是订单实体的内容，会造成在消费端进行消费时如果需要去验证订单是否存在时可能出现订单不存在。
3、消息发送可以考虑异步发送。
方案二：
由于存在上述问题，在MQ不支持事务消息的前提条件下，可以采用下面的方式进行优化。
然后在控制器层，使用异步发送，将消息发送，并在消息发送成功后，更新待发送状态为已发送。
然后通过定时任务，扫描待发送，结合创建时间的记录（小于当前时间5分钟的消息待发送记录），进行消息发送。
方案弊端：
1、消息有可能重复发送，但在消费端可以通过唯一业务编号来进行去重设计。
2、实现过于复杂，为了避免 极端情况下的消息丢失，需要使用定时任务。
方案三：基于RocketMQ4.3版本事务消息
额外需要实现事务会查监听器：TransactionListener，其实例代码：
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.concurrent.ConcurrentHashMap;
@SuppressWarnings("unused")
public class OrderTransactionListenerImpl implements TransactionListener
private ConcurrentHashMap<String, Integer> countHashMap = new ConcurrentHashMap<>();
private final static int MAX_COUNT = 5;
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg)
//
String bizUniNo = msg.getUserProperty("bizUniNo"); // 从消息中获取业务唯一ID。
// 将bizUniNo入库，表名：t_message_transaction,表结构 bizUniNo(主键),业务类型。
return LocalTransactionState.UNKNOW;
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg)
Integer status = 0;
// 从数据库查查询t_message_transaction表，如果该表中存在记录，则提交，
String bizUniNo = msg.getUserProperty("bizUniNo"); // 从消息中获取业务唯一ID。
// 然后t_message_transaction 表，是否存在bizUniNo，如果存在，则返回COMMIT_MESSAGE，
// 不存在，则记录查询次数，未超过次数，返回UNKNOW，超过次数，返回ROLLBACK_MESSAGE
if(query(bizUniNo) > 0 )
return LocalTransactionState.COMMIT_MESSAGE;
return rollBackOrUnown(bizUniNo);
public int query(String bizUniNo)
return 1; //select count(1) from t_message_transaction a where a.biz_uni_no=#bizUniNo
public LocalTransactionState rollBackOrUnown(String bizUniNo)
Integer num = countHashMap.get(bizUniNo);
if(num != null && ++num > MAX_COUNT)
countHashMap.remove(bizUniNo);
return LocalTransactionState.ROLLBACK_MESSAGE;
if(num == null)
num = new Integer(1);
countHashMap.put(bizUniNo, num);
return LocalTransactionState.UNKNOW;
TransactionListener 实现要点：
executeLocalTransaction：
该方法，主要是设置本地事务状态，该方法与业务方代码在一个事务中，例如OrderServer#createMap中，只要本地事务提交成功，该方法也会提交成功。
故在这里，主要是t_message_transaction添加一条记录，在事务会查时，如果存在记录，就认为是该消息需要提交。
checkLocalTransaction：
该方法主要是告知RocketMQ消息是否需要提交还是回滚，如果本地事务表（t_message_transaction）存在记录，则认为提交，如果不存在，可以设置会查次数，如果指定次数内还是未查到消息，则回滚，否则返回未知，rocketmq会按一定的频率回查事务，当然回查次数也有限制，默认为5次，可配置。
温馨提示：业务的操作与本地事务消息表都需要在 OrderTransactionListenerImpl 的 executeLocalTransaction 方法中执行，确保是一个事务。
见文如面，我是威哥，热衷于成体系剖析JAVA主流中间件，关注公众号『中间件兴趣圈』，回复专栏可获取成体系专栏导航，回复资料可以获取笔者的学习思维导图。
rocketmq入门到入土事务消息&顺序消息(代码片段)
接上一篇：RocketMQ入门到入土（一）新手也能看懂的原理和实战！一、事务消息的由来1、案例引用官方的购物案例：小明购买一个100元的东西，账户扣款100元的同时需要保证在下游的积分系统给小明这个账号增加100积分。账号系... 查看详情
rocketmq事务消息入门介绍(代码片段)
说明周五的时候发了篇：Rocketmq4.3支持事务啦！！！，趁着周末的时候把相关内容看了下，下面的主要内容就是关于RocketMQ事务相关内容介绍了。说明：今天这篇仅仅是入门介绍，并没有涉及到很多细节，先把大概流程说明白，... 查看详情
rocketmq（09）——发送事务消息(代码片段)
发送事务消息RocketMQ支持发送事务消息，它的事务消息是基于二阶段提交机制实现的。当发送的消息是事务消息时，只有对应的消息被提交了才能被消费者进行消费。发送事务消息时生产者需要使用TransactionMQProducer，它还需要指... 查看详情
rocketmq事务消息篇之事务消息的使用(代码片段)
前言在RocketMQ事务消息篇（一）之事务消息的介绍里对RocketMQ的事务消息作了相关说明，本文提供一些基本的开发示例。java示例依赖<dependencies><dependency><groupId>org.apache.rocketmq</groupId><artifactId>ro 查看详情
rocketmq（09）——发送事务消息(代码片段)
发送事务消息RocketMQ支持发送事务消息，它的事务消息是基于二阶段提交机制实现的。当发送的消息是事务消息时，只有对应的消息被提交了才能被消费者进行消费。发送事务消息时生产者需要使用TransactionMQProducer，... 查看详情
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实现事务消息方案(代码片段)
RocketMQ是一个来自阿里巴巴的分布式消息中间件，于2012年开源，并在2017年正式成为Apache顶级项目。据了解，包括阿里云上的消息产品以及收购的子公司在内，阿里集团的消息产品全线都运行在RocketMQ之上，并且最近几年的双十一... 查看详情
rocketmq使用事务消息(代码片段)
目录说明原理事务消息处理流程生产端消费端说明事务消息：1、不支持延时消息和批量消息2、如果消息没有及时提交，默认check15次，可以通过Broker的transactionCheckMax参数配置次数。如果超时15次依然没有得到明确结果... 查看详情
rocketmq源码分析之从官方示例窥探：rocketmq事务消息实现基本思想(代码片段)
RocketMQ4.3.0版本开始支持事务消息，后续分享将开始将剖析事务消息的实现原理。首先从官方给出的Demo实例入手，以此通往RocketMQ事务消息的世界中。官方版本未发布之前，从apacherocketmq第一个版本上线后，代码中存在与事务消息... 查看详情
rocketmq源码分析之rocketmq事务消息实现原理中篇----事务消息状态回查(代码片段)
上节已经梳理了RocketMQ发送事务消息的流程（基于二阶段提交），本节将继续深入学习事务状态消息回查，我们知道，第一次提交到消息服务器时消息的主题被替换为RMQ_SYS_TRANS_HALF_TOPIC，本地事务执行完后如果返回本地事务状态... 查看详情
rocketmq源码分析之rocketmq事务消息实现原理中篇----事务消息状态回查(代码片段)
上节已经梳理了RocketMQ发送事务消息的流程（基于二阶段提交），本节将继续深入学习事务状态消息回查，我们知道，第一次提交到消息服务器时消息的主题被替换为RMQ_SYS_TRANS_HALF_TOPIC，本地事务执行完后如果返回本地事务状态... 查看详情
七.rocketmq极简入门-rocketmq事务消息(代码片段)
概述如果业务只涉及到一个数据库的写操作，我们只需要保证这一个事物的提交和回滚，这种事务管理叫传统事物或本地事务，如果业务涉及到多个数据库(多个服务)的写操作，我们需要保证多个数据库同时提交... 查看详情
rocketmq的分布式事务机制（事务消息）(代码片段)
详细介绍了RocketMQ的事务消息机制，RocketMQ的事务消息可以用于实现基于可靠消息的最终一致性的分布式事务。文章目录1事务消息简要流程2一阶段半消息不可见的设计3二阶段Commit和Rollback操作4Op消息的设计5Commit消息变得可见... 查看详情
rocketmq事务消息机制(代码片段)
RocketMQ提供了事务消息，通过事务消息就能达到分布式事务的最终一致，从而实现了可靠消息服务。一、事务消息的实现步骤事务消息发送步骤：1.发送方将半事务消息发送至RocketMQ服务端。2.RocketMQ服务端将消息持久... 查看详情
springboot实战项目整合阿里云rocketmq消息队列实现发送普通消息，延时消息(代码片段)
原文地址：Springboot实战项目整合阿里云RocketMQ消息队列实现发送普通消息，延时消息--附代码-学不会丶-博客园一.为什么选择RocketMQ消息队列？（可跳过看三的整合代码实例）首先RocketMQ是阿里巴巴自研出来的&#... 查看详情