在单体应用中，一个@Transactional注解就能搞定事务，可一旦拆成微服务，订单服务扣库存、账户服务扣余额这类跨库操作就无法依赖数据库事务了。硬扛着用强一致性方案（如两阶段提交）性能开销大、实现复杂度也高，绝大多数业务其实只需要最终一致性就够用。RocketMQ 的事务消息正是为这个场景设计的：先发半消息，执行本地事务，再根据本地事务结果决定提交或回滚消息，配合消费端的重试与幂等机制，实现整套最终一致性链路。这篇文章把原理和代码完整过一遍。
一、事务消息的底层原理：半消息与回查机制
要理解 RocketMQ 事务消息，关键是抓住三个阶段。第一阶段是发送半消息（Half Message）：生产者先把消息发到 Broker，但这条消息对消费者完全不可见，被存在一个特殊的系统 Topic 里。第二阶段是执行本地事务：生产者发送半消息成功后，回调本地的执行器方法，比如在订单库里插入一条记录。第三阶段是根据本地事务结果做二次确认：本地事务成功就提交消息（Commit），让消费者可以消费；失败则回滚（Rollback），这条半消息会被删除。
这里有个隐藏的问题：如果生产者在执行完本地事务之后、发送二次确认之前，进程挂了怎么办？Broker 收不到确认，这条半消息既不能提交也不能删除。这就轮到回查机制登场了。Broker 会定期扫描长时间未确认的半消息（默认从 6 秒后开始，最多回查 15 次），主动反查生产者：这条消息对应的本地事务到底执行成功没有？生产者需要在回查方法里根据业务状态返回 Commit 或 Rollback。
正因为存在回查，本地事务记录就必须落库可查。常见做法是建一张本地事务表，记录消息 ID、事务状态、业务唯一键等信息。回查时查这张表就能给出准确答案，而不是靠内存变量。如果 15 次回查都没结果，Broker 会默认丢弃这条消息（DefaultMQProducerImpl 中默认回滚），并在日志中留下痕迹，这种情况需要配合监控告警及时发现。
二、Spring Boot 整合步骤：从依赖到生产者代码
整合用官方的 rocketmq-spring-boot-starter 最省事。以 Spring Boot 2.7.x 和 rocketmq-spring-boot-starter 2.2.3 为例，先引入依赖：
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.2.3</version>
</dependency>接着在 application.yml 中配置 NameServer 地址和生产者组。注意生产者组名称必须全局唯一，多个服务复用同一个组名会导致事务回查路由到错误的生产者实例：
rocketmq:
name-server: 192.168.0.1:9876
producer:
group: order-tx-producer-group
send-message-timeout: 5000然后编写核心的事务监听器。这个类要实现 RocketMQListener 接口里的 executeLocalTransaction 和 checkLocalTransaction 两个方法，前者负责执行本地事务，后者负责应对 Broker 的回查。用 RocketMQTransactionListener 注解标记并指定事务生产者的 Bean 名称：
@RocketMQTransactionListener
public class OrderTxListener implements RocketMQLocalTransactionListener {
@Resource
private OrderMapper orderMapper;
@Resource
private LocalTxRecordMapper txRecordMapper;
@Override
public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
String orderId = (String) arg;
try {
// 本地事务：保存订单并记录事务状态表，同一本地事务内保证原子性
Order order = buildOrder(orderId);
orderMapper.insert(order);
txRecordMapper.insert(buildTxRecord(orderId, msg.getHeaders().get("JMS_MESSAGE_ID", String.class), "COMMIT"));
return RocketMQLocalTransactionState.COMMIT;
} catch (Exception e) {
// 本地事务失败，回滚半消息
return RocketMQLocalTransactionState.ROLLBACK;
}
}
@Override
public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
String orderId = (String) msg.getHeaders().get("orderId");
// 回查时根据本地事务表判断状态
LocalTxRecord record = txRecordMapper.selectByOrderId(orderId);
if (record == null) {
// 可能本地事务还没执行完，返回 UNKNOWN 等待下次回查
return RocketMQLocalTransactionState.UNKNOWN;
}
return "COMMIT".equals(record.getStatus())
? RocketMQLocalTransactionState.COMMIT
: RocketMQLocalTransactionState.ROLLBACK;
}
}发送消息的代码更简单，直接注入 RocketMQTemplate 调用 sendMessageInTransaction 方法。第一个参数是目标 Topic，arg 参数会原样传递到 executeLocalTransaction 中，一般用来传业务主键：
@Service
public class OrderService {
@Resource
private RocketMQTemplate rocketMQTemplate;
@Transactional(rollbackFor = Exception.class)
public void createOrder(String orderId) {
// 发送半消息，arg 传入订单号
rocketMQTemplate.sendMessageInTransaction("order-tx-topic", MessageBuilder
.withPayload(new OrderDTO(orderId))
.setHeader("orderId", orderId)
.build(), orderId);
}
}三、消费端实现与幂等处理
消费端用 @RocketMQMessageListener 注解声明一个监听类即可。需要重点关注三件事：消费组命名、异常时的重试行为、以及幂等保障。RocketMQ 对消费失败的消息默认重试 16 次，仍失败则进入死信队列，所以业务代码里不要吞掉异常，该抛就抛，让重试机制接管：
@Component
@RocketMQMessageListener(
topic = "order-tx-topic",
consumerGroup = "stock-consumer-group",
maxReconsumeTimes = 5)
public class StockConsumer implements RocketMQListener<MessageExt> {
@Resource
private StockService stockService;
@Override
public void onMessage(MessageExt message) {
String orderId = message.getKeys();
String msgId = message.getMsgId();
// 幂等校验：先查消费记录表，已消费则直接返回
if (consumeRecordService.exists(msgId, orderId)) {
return;
}
stockService.deduct(orderId);
// 记录消费状态，与扣减操作在同一本地事务中完成
consumeRecordService.save(msgId, orderId);
}
}幂等为什么必不可少？因为消息投递语义是至少一次（At Least Once），网络抖动、消费超时都会导致同一条消息被重复投递。幂等实现有多种方案：数据库唯一索引（以消息 ID 建唯一键）、Redis 的 SETNX 加过期时间、或者状态机检查（已扣减的订单再扣直接跳过）。生产环境建议用数据库唯一索引兜底，因为它和业务操作在同一事务里，天然原子；Redis 方案则要额外处理缓存与数据库的一致性问题，复杂度更高。
四、生产环境常见坑与优化建议
第一个坑是回查方法里返回 UNKNOWN 的滥用。UNKNOWN 意味着等待下一次回查，但如果业务上无法判断状态就一直返回 UNKNOWN，15 次之后消息被丢弃，数据就静默丢失了。正确做法是回查逻辑必须能查到确定结果，本地事务表是标配。第二个坑是事务监听器里的 Bean 注入问题，早期版本 starter 要求监听器不能是纯代理对象，遇到注入失败可以升级 starter 版本或调整 AOP 配置。
第三个坑是消费端阻塞。事务消息只保证生产端与 Broker 的一致性，消费端处理慢照样会堆积。要给消费逻辑设置合理的超时，批量拉取参数（consumeMessageBatchMaxSize）按吞吐调优，并监控消费者组的堆积量指标。第四个坑是发送顺序问题：sendMessageInTransaction 本身不保证顺序，若下游依赖顺序消费，需要用顺序消息（MessageQueueSelector 指定队列）配合，但事务消息和顺序消息叠加会显著增加复杂度，非必要不混用。
最后提一下监控。事务消息的关键指标包括半消息数量、回查次数、回查超时丢弃数，这些都可以通过 RocketMQ Dashboard 或者消息轨迹（traceTopic）功能观测到。建议把回查丢弃量接入告警，因为每一次丢弃都意味着一笔可能不一致的业务数据，需要人工介入核对补偿。整体方案跑通后，订单、库存、账户三个服务各管各的本地事务，通过消息串联，既保证了最终一致性，又保留了各服务的自治能力，这就是事务消息在微服务架构中的核心价值。
Spring BootRocketMQ事务消息修改时间：2026-09-13 23:47:11