前言：分布式系统的“一致性”痛点与解决方案​

在上一篇文章中，我们掌握了 RocketMQ 消息全链路追踪与全方位性能优化技巧，确保了集群在高并发场景下的稳定高效运行。但在分布式系统中，除了“稳定”和“高效”，还有一个核心痛点亟待解决——分布式事务一致性

试想这样的生产场景：电商系统中，用户下单后，需要完成两个核心操作：一是扣减商品库存（库存服务），二是创建订单记录（订单服务）。理想情况下，两个操作要么同时成功，要么同时失败，确保数据一致性；但实际情况中，可能出现“库存扣减成功，但订单创建失败”的异常——此时库存已减少，订单却未生成，导致数据不一致，进而引发超卖、用户投诉等问题；反之，“订单创建成功，库存扣减失败”，则会导致无货可发，同样造成业务损失。

这类问题的本质，是分布式系统中“跨服务业务操作”的原子性无法保证——传统的本地事务（如数据库事务）只能作用于单个服务，无法跨服务协调。而 RocketMQ 提供的分布式事务消息，正是解决这一痛点的核心方案：它通过“半事务消息+事务确认”的机制，将跨服务的业务操作与消息发送绑定，确保“业务操作与消息发送”要么同时成功、要么同时失败，最终实现分布式系统的数据一致性。​

本篇作为高级篇的第三篇，将从分布式事务的核心概念入手，深度解析 RocketMQ 分布式事务消息的实现原理，手把手教你搭建分布式事务消息环境、编写实战代码，同时讲解事务回滚、异常处理、幂等性保障等生产级关键细节，结合电商下单、支付回调等典型场景，帮你彻底掌握 RocketMQ 分布式事务消息的应用，解决分布式系统中的一致性难题。

前置要求：​
已掌握 RocketMQ 高可用集群部署（Dledger 模式）、消息追踪与性能优化方法​
了解分布式事务的基本概念（如 CAP 理论、最终一致性、原子性）​
具备 Java 项目开发、Spring Boot 集成 RocketMQ 基础，熟悉数据库事务操作​
已搭建 RocketMQ 官方控制台，可查看消息状态、事务状态等信息

一、分布式事务核心认知：为什么需要事务消息？​

在讲解 RocketMQ 事务消息之前，我们先明确分布式事务的核心痛点、常见解决方案，以及 RocketMQ 事务消息的定位，帮助大家理解其设计初衷和应用场景。

1.1 分布式事务的核心痛点​
分布式系统中，多个服务分散部署、独立存储数据，跨服务的业务操作无法通过本地事务保证原子性，主要存在以下3个核心痛点：​
数据不一致：多个服务的操作无法同步成功或失败，导致数据偏差（如库存扣减成功、订单创建失败）。​
异常难以处理：某个服务操作失败后，无法快速回滚其他已成功的操作，且异常排查难度大（需跨服务核对日志）。​
性能与一致性平衡难：追求强一致性会导致服务耦合度高、性能下降；追求高性能又可能牺牲数据一致性。

1.2 分布式事务常见解决方案对比​
目前分布式事务有多种解决方案，各有优劣，适用于不同场景，我们重点对比与 RocketMQ 事务消息相关的方案：

解决方案	解决方案	优势	劣势
2PC/3PC	分准备、提交/回滚两阶段，由协调者统一调度	强一致性，实现简单	协调者单点故障、阻塞严重、性能差
TCC 补偿	Try（尝试）、Confirm（确认）、Cancel（取消）三阶段，手动编写补偿逻辑	性能好、灵活性高，支持复杂业务	开发成本高、需手动处理补偿和幂等
RocketMQ 事务消息	半事务消息+事务确认+定时回查，基于消息队列实现最终一致性	开发成本低、无阻塞、性能好，无需手动补偿	仅支持最终一致性，不适合强一致性

选型建议：大多数分布式业务（如电商下单、支付回调、物流通知），无需强一致性，追求“最终一致性”即可，优先选择 RocketMQ 事务消息——开发成本低、性能好，且能依托 RocketMQ 集群的高可用性，减少系统复杂度。

1.3 RocketMQ 事务消息的核心定位
RocketMQ 事务消息的核心目标是：保证“本地业务操作”与“消息发送”的原子性，即本地业务操作成功，则消息最终能被消费者消费；本地业务操作失败，则消息不会被消费者消费（或被回滚），从而实现跨服务业务的最终一致性。

注意：RocketMQ 事务消息实现的是最终一致性，而非强一致性——消息从发送到被消费者消费，存在一定延迟，但最终会达到数据一致状态，适合能接受短暂不一致的业务场景（如电商下单、积分兑换）。

二、RocketMQ 事务消息原理：半事务消息+事务确认机制

RocketMQ 事务消息的核心机制是“半事务消息”和“事务确认”，配合“定时回查”机制，解决分布式事务中的一致性问题，整个流程可分为3个核心阶段：发送半事务消息、执行本地事务、确认事务状态。

2.1 核心概念解析

• 半事务消息：消息发送到 Broker 后，Broker 会先存储该消息，但不会立即将其投递给消费者（处于“不可消费”状态），这种状态的消息称为半事务消息。只有当生产者确认“本地事务执行成功”后，Broker 才会将消息标记为“可消费”，投递给消费者；若确认“本地事务执行失败”，Broker 会删除该消息，不会投递。

• 事务确认（Commit/Rollback）：生产者执行完本地事务后，向 Broker 发送确认指令——Commit（确认成功）或 Rollback（确认失败）。Broker 收到 Commit 指令，将半事务消息标记为可消费；收到 Rollback 指令，删除半事务消息。

• 定时回查（事务状态回查）：若生产者执行本地事务后，因网络异常、服务宕机等原因，未向 Broker 发送确认指令，Broker 会定期（默认 1 分钟）向生产者发起事务状态回查，询问本地事务执行结果，进而决定消息是否投递或删除。这是 RocketMQ 事务消息可靠性的核心保障，避免因异常导致数据不一致。

• 事务消息生产者（TransactionProducer）：与普通生产者不同，事务消息生产者需要实现事务监听器（TransactionListener），用于执行本地事务和处理 Broker 的回查请求。

2.2 完整执行流程（图文解析）
以“电商下单”场景（订单服务+库存服务）为例，RocketMQ 事务消息的完整执行流程如下，确保“订单创建”与“库存扣减”的最终一致性：

1. 步骤1：发送半事务消息：订单服务（生产者）向 RocketMQ Broker 发送半事务消息（消息内容：“扣减商品库存”），Broker 存储消息后，返回“消息发送成功”（此时消息不可消费）。

2. 步骤2：执行本地事务：订单服务收到 Broker 的“消息发送成功”响应后，执行本地事务——创建订单记录（写入订单数据库）。

3. 步骤3：事务确认：

• 若本地事务执行成功（订单创建成功），订单服务向 Broker 发送 Commit 指令，Broker 将半事务消息标记为“可消费”，投递给库存服务（消费者）。

• 若本地事务执行失败（订单创建失败），订单服务向 Broker 发送 Rollback 指令，Broker 删除半事务消息，库存服务不会收到任何消息。

4. 步骤4：消费者消费消息：库存服务（消费者）收到 Broker 投递的消息后，执行本地业务——扣减商品库存，完成整个跨服务业务流程。

5. 步骤5：定时回查（异常处理）：若步骤3中，订单服务因网络异常、服务宕机等原因，未向 Broker 发送 Commit/Rollback 指令，Broker 会定期向订单服务发起事务回查，询问本地事务执行结果：

• 若回查发现本地事务执行成功，Broker 自动将消息标记为可消费，投递到库存服务。

• 若回查发现本地事务执行失败，Broker 自动删除半事务消息。

• 若回查多次（默认 15 次）仍无法获取事务状态，Broker 会将消息标记为“异常消息”，存入死信队列，由人工干预处理。

核心亮点：
整个流程中，“半事务消息”是关键——它确保了“消息发送”与“本地事务”的绑定：只有消息发送成功，才会执行本地事务；只有本地事务执行成功，消息才会被投递。即使出现异常，通过定时回查机制，也能确保消息状态最终一致，避免数据偏差。

2.3 关键参数配置（Broker 端）
RocketMQ 事务消息的相关参数默认已配置，无需修改即可使用，若需根据业务场景调整，可修改 Broker 配置文件（broker.conf），核心参数如下：

# 事务消息回查间隔时间（毫秒），默认 60000ms（1分钟）
transactionCheckInterval = 60000

# 事务消息最大回查次数，默认 15 次
transactionCheckMax = 15

# 事务消息死信队列前缀，默认 %DLQ%
dlqTopicName = %DLQ%

# 是否开启事务消息功能，默认 true（开启）
transactionMessageEnable = true

说明：若事务消息回查 15 次后，仍无法获取事务状态，Broker 会将该消息发送到对应生产者组的死信队列（%DLQ%+生产者组名），避免消息一直占用资源，后续可通过死信队列进行人工排查和处理。

三、实操：RocketMQ 分布式事务消息实战（Spring Boot 集成）

本次实操以“电商下单”为核心场景，基于 Spring Boot 2.7.x + RocketMQ 4.9.0，实现“订单创建”与“库存扣减”的分布式事务一致性，步骤可直接落地到生产环境。

实操架构：订单服务（事务消息生产者）→ RocketMQ Broker → 库存服务（消息消费者），确保“订单创建成功”与“库存扣减成功”的最终一致性。

3.1 环境准备

1. 已部署 RocketMQ 高可用集群（Dledger 模式），确保 Broker 开启事务消息功能（默认开启）。

2. 创建两个 Spring Boot 项目：order-service（订单服务，生产者）、inventory-service（库存服务，消费者）。

3. 引入 RocketMQ 依赖（两个项目均需引入）：

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.3</version>
</dependency>
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>8.0.30</version>
</dependency>
<dependency>
    <groupId>org.mybatis.spring.boot</groupId>
    <artifactId>mybatis-spring-boot-starter</artifactId>
    <version>2.2.2</version>
</dependency>

3.2 订单服务（事务消息生产者）开发
订单服务作为事务消息生产者，核心功能：发送半事务消息、执行本地事务（创建订单）、处理 Broker 事务回查，步骤如下：

3.2.1 配置文件（application.yml）

spring:
  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://192.168.1.105:3306/order_db?useUnicode=true&characterEncoding=utf-8&serverTimezone=GMT%2B8
    username: root
    password: 123456
mybatis:
  mapper-locations: classpath:mapper/*.xml
  type-aliases-package: com.example.orderservice.entity
rocketmq:
  name-server: 192.168.1.101:9876;192.168.1.102:9876;192.168.1.103:9876
  producer:
    group: order-transaction-producer-group # 事务消息生产者组
    send-message-timeout: 3000 # 消息发送超时时间
    retry-times-when-send-failed: 1 # 发送失败重试次数

3.2.2 实体类与数据库设计

1. 订单表（order_db.order）：

CREATE TABLE `order` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '订单ID',
  `order_no` varchar(50) NOT NULL COMMENT '订单编号',
  `user_id` bigint(20) NOT NULL COMMENT '用户ID',
  `product_id` bigint(20) NOT NULL COMMENT '商品ID',
  `product_num` int(11) NOT NULL COMMENT '商品数量',
  `order_status` tinyint(4) NOT NULL DEFAULT 0 COMMENT '订单状态：0-待支付，1-已支付，2-已取消',
  `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  PRIMARY KEY (`id`),
  UNIQUE KEY `idx_order_no` (`order_no`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='订单表';

2. 实体类（Order.java）：

package com.example.orderservice.entity;

import lombok.Data;
import java.util.Date;

@Data
public class Order {
    private Long id;
    private String orderNo;
    private Long userId;
    private Long productId;
    private Integer productNum;
    private Integer orderStatus;
    private Date createTime;
}

3.2.3 事务监听器（核心）
事务消息生产者必须实现 TransactionListener 接口，重写两个方法：

• executeLocalTransaction：执行本地事务（创建订单），返回事务状态（COMMIT_MESSAGE、ROLLBACK_MESSAGE、UNKNOWN）。

• checkLocalTransaction：处理 Broker 的事务回查，返回事务状态。

package com.example.orderservice.listener;

import com.example.orderservice.entity.Order;
import com.example.orderservice.service.OrderService;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.MessageExt;

@Component
public class OrderTransactionListener implements TransactionListener {

    @Autowired
    private OrderService orderService;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    // 1. 执行本地事务
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            // 从消息中获取订单信息（arg 为发送消息时传入的订单对象）
            Order order = (Order) arg;
            // 执行本地事务：创建订单（数据库事务）
            boolean createResult = orderService.createOrder(order);
            // 本地事务执行成功，返回 COMMIT，通知 Broker 投递消息
            if (createResult) {
                return LocalTransactionState.COMMIT_MESSAGE;
            } else {
                // 本地事务执行失败，返回 ROLLBACK，通知 Broker 删除消息
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        } catch (Exception e) {
            // 出现异常，返回 UNKNOWN，等待 Broker 回查
            e.printStackTrace();
            return LocalTransactionState.UNKNOWN;
        }
    }

    // 2. 处理 Broker 事务回查
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        try {
            // 从消息中获取订单编号（消息体中存储订单编号，用于查询订单状态）
            String orderNo = new String(msg.getBody());
            //  查询本地订单状态：判断订单是否创建成功
            Order order = orderService.getOrderByOrderNo(orderNo);
            if (order != null && order.getOrderStatus() == 0) {
                // 订单已创建，本地事务执行成功，返回 COMMIT
                return LocalTransactionState.COMMIT_MESSAGE;
            } else if (order == null) {
                // 订单未创建，本地事务执行失败，返回 ROLLBACK
                return LocalTransactionState.ROLLBACK_MESSAGE;
            } else {
                // 订单状态异常，返回 UNKNOWN，继续回查
                return LocalTransactionState.UNKNOWN;
            }
        } catch (Exception e) {
            e.printStackTrace();
            return LocalTransactionState.UNKNOWN;
        }
    }
}

3.2.4 订单服务与消息发送接口

1. 订单服务（OrderService.java）：实现订单创建（本地事务）和订单查询功能，确保本地事务的原子性。

package com.example.orderservice.service;

import com.example.orderservice.entity.Order;
import com.example.orderservice.mapper.OrderMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.UUID;

@Service
public class OrderService {

    @Autowired
    private OrderMapper orderMapper;

    // 本地事务：创建订单，添加 @Transactional 确保本地事务原子性
    @Transactional(rollbackFor = Exception.class)
    public boolean createOrder(Order order) {
        // 生成唯一订单编号
        order.setOrderNo(UUID.randomUUID().toString().replace("-", ""));
        order.setOrderStatus(0); // 待支付状态
        // 插入订单记录
        int insert = orderMapper.insertOrder(order);
        // 若插入成功，返回 true；失败返回 false
        return insert > 0;
    }

    // 查询订单（用于事务回查）
    public Order getOrderByOrderNo(String orderNo) {
        return orderMapper.selectOrderByOrderNo(orderNo);
    }
}

2. 消息发送接口（OrderController.java）：提供下单接口，发送半事务消息。

package com.example.orderservice.controller;

import com.example.orderservice.entity.Order;
import com.example.orderservice.service.OrderService;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class OrderController {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private OrderService orderService;

    // 下单接口：发送半事务消息
    @PostMapping("/createOrder")
    public String createOrder(@RequestBody Order order) {
        try {
            // 1. 发送半事务消息：topic 为 inventory-topic，消息体为订单编号，arg 为订单对象（用于本地事务）
            rocketMQTemplate.sendMessageInTransaction(
                "inventory-topic", // 消息主题（库存服务订阅的主题）
                order.getOrderNo(), // 消息体（存储订单编号，用于库存服务关联订单）
                order // 传入本地事务的参数（订单对象）
            );
            return "下单请求提交成功，订单正在创建中";
        } catch (Exception e) {
            e.printStackTrace();
            return "下单失败，请重试";
        }
    }
}

3.3 库存服务（消息消费者）开发
库存服务作为消费者，核心功能：接收 Broker 投递的消息（扣减库存），处理消息消费异常，确保库存扣减的原子性，步骤如下：

3.3.1 配置文件（application.yml）

spring:
  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://192.168.1.105:3306/inventory_db?useUnicode=true&characterEncoding=utf-8&serverTimezone=GMT%2B8
    username: root
    password: 123456
mybatis:
  mapper-locations: classpath:mapper/*.xml
  type-aliases-package: com.example.inventoryservice.entity
rocketmq:
  name-server: 192.168.1.101:9876;192.168.1.102:9876;192.168.1.103:9876
  consumer:
    group: inventory-consumer-group # 消费者组
    topic: inventory-topic # 订阅的主题
    message-model: CLUSTERING # 集群模式（默认）
    consume-thread-min: 8
    consume-thread-max: 16

3.3.2 实体类与数据库设计

1. 库存表（inventory_db.inventory）：

CREATE TABLE `inventory` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '库存ID',
  `product_id` bigint(20) NOT NULL COMMENT '商品ID',
  `stock_num` int(11) NOT NULL DEFAULT 0 COMMENT '商品库存数量',
  `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  PRIMARY KEY (`id`),
  UNIQUE KEY `idx_product_id` (`product_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='库存表';

2. 实体类（Inventory.java）：

package com.example.inventoryservice.entity;

import lombok.Data;
import java.util.Date;

@Data
public class Inventory {
    private Long id;
    private Long productId;
    private Integer stockNum;
    private Date updateTime;
}

3.3.3 库存服务与消息监听

1. 库存服务（InventoryService.java）：实现库存扣减功能，确保本地事务原子性。

package com.example.inventoryservice.service;

import com.example.inventoryservice.entity.Inventory;
import com.example.inventoryservice.mapper.InventoryMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
public class InventoryService {

    @Autowired
    private InventoryMapper inventoryMapper;

    // 扣减库存（本地事务）
    @Transactional(rollbackFor = Exception.class)
    public boolean deductStock(Long productId, Integer productNum) {
        // 1. 查询商品库存
        Inventory inventory = inventoryMapper.selectByProductId(productId);
        if (inventory == null || inventory.getStockNum() < productNum) {
            // 库存不足，返回失败
            return false;
        }
        // 2. 扣减库存
        int update = inventoryMapper.deductStock(productId, productNum);
        return update > 0;
    }
}

2. 消息监听器（InventoryMessageListener.java）：接收消息，调用库存扣减服务，处理消费异常。

package com.example.inventoryservice.listener;

import com.example.inventoryservice.service.InventoryService;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

// 订阅 inventory-topic 主题，消费者组为 inventory-consumer-group
@RocketMQMessageListener(topic = "inventory-topic", consumerGroup = "inventory-consumer-group")
@Component
public class InventoryMessageListener implements RocketMQListener<String> {

    @Autowired
    private InventoryService inventoryService;

    // 接收消息，扣减库存
    @Override
    public void onMessage(String orderNo) {
        try {
            // 这里简化处理：假设从订单服务获取商品ID和数量（实际场景中可通过订单编号查询订单信息）
            Long productId = 1001L; // 示例商品ID
            Integer productNum = 1; // 示例商品数量
            // 扣减库存
            boolean deductResult = inventoryService.deductStock(productId, productNum);
            if (!deductResult) {
                // 库存不足，抛出异常，让 RocketMQ 重试消费
                throw new RuntimeException("库存不足，扣减失败");
            }
            System.out.println("订单" + orderNo + "：库存扣减成功");
        } catch (Exception e) {
            e.printStackTrace();
            // 消费失败，抛出异常，RocketMQ 会自动重试（默认重试 16 次）
            throw new RuntimeException("库存扣减异常，触发重试", e);
        }
    }
}

3.4 测试验证（确保一致性）
部署订单服务和库存服务后，通过 Postman 调用下单接口（/createOrder），测试3种场景，验证分布式事务一致性：

场景1：本地事务执行成功
请求参数：{“userId”:1001,“productId”:1001,“productNum”:1}

预期结果：

• 订单表中新增一条订单记录（order_status=0）。

• RocketMQ 消息被标记为可消费，库存服务收到消息，扣减对应商品库存。

• 库存表中商品1001的库存减少1。

场景2：本地事务执行失败

模拟订单创建失败（如手动抛出异常），请求参数同上。

预期结果：

• 订单表中无新增记录（本地事务回滚）。

• RocketMQ 收到 Rollback 指令，删除半事务消息，库存服务未收到消息，库存无变化。

场景3：网络异常，未发送事务确认

模拟订单创建成功，但网络异常，未向 Broker 发送 Commit 指令。

预期结果：

• 订单表中新增订单记录。

• Broker 1分钟后发起事务回查，查询到订单已创建，自动将消息标记为可消费。

• 库存服务收到消息，扣减库存，最终数据一致。

四、生产级关键细节：异常处理与幂等性保障

在实际生产场景中，分布式事务消息可能面临各种异常（如消息重试、服务宕机、消息重复消费），若不处理这些异常，依然会导致数据不一致。以下是生产级关键细节，必须重点关注：

4.1 消息重复消费与幂等性处理

由于网络异常、Broker 回查、消息重试等原因，消费者可能会收到重复的消息（如库存服务多次收到同一订单的扣减库存消息），若不处理，会导致库存重复扣减（如订单1001的库存被扣减2次），引发数据不一致。

解决方案：实现消费幂等性——确保同一消息被消费多次，结果依然一致，常用方案如下：

• 基于唯一标识去重：

  • 消息体中携带唯一标识（如订单编号、消息 ID）。

  • 消费者消费消息前，先查询“消费记录表”（存储已消费的消息唯一标识），若该消息已消费，直接返回成功；若未消费，执行消费逻辑，消费完成后写入消费记录表。

• 基于业务逻辑幂等：在库存扣减逻辑中，增加判断——根据订单编号查询“库存扣减记录”，若该订单已扣减过库存，直接返回成功；若未扣减，执行扣减逻辑。

// 优化库存扣减逻辑，实现幂等性（基于订单编号）
@Transactional(rollbackFor = Exception.class)
public boolean deductStock(Long productId, Integer productNum, String orderNo) {
    // 1. 查询该订单是否已扣减过库存（假设存在库存扣减记录表 inventory_deduct）
    Integer count = inventoryMapper.selectDeductByOrderNo(orderNo);
    if (count > 0) {
        // 已扣减过，直接返回成功（幂等处理）
        return true;
    }
    // 2. 查询商品库存
    Inventory inventory = inventoryMapper.selectByProductId(productId);
    if (inventory == null || inventory.getStockNum() < productNum) {
        return false;
    }
    // 3. 扣减库存
    int update = inventoryMapper.deductStock(productId, productNum);
    if (update > 0) {
        // 4. 写入库存扣减记录，用于幂等校验
        inventoryMapper.insertDeductRecord(orderNo, productId, productNum);
        return true;
    }
    return false;
}

4.2 事务回查异常处理
若生产者服务宕机、数据库异常，导致 Broker 多次回查仍无法获取事务状态，消息会被存入死信队列，此时需要人工干预处理，避免消息丢失或数据不一致：

1. 死信队列监控：通过 RocketMQ 控制台监控死信队列（%DLQ%+生产者组名），若有异常消息，及时告警。

2. 人工排查与处理：查询死信消息的内容（如订单编号），核对本地事务状态（订单是否创建成功），手动决定是“重新投递消息”还是“删除消息”。

3. 重试机制优化：调整事务回查间隔（transactionCheckInterval）和最大回查次数（transactionCheckMax），根据业务场景优化，避免频繁回查占用资源。

4.3 本地事务与消息发送的原子性保障
核心原则：只有半事务消息发送成功，才执行本地事务；只有本地事务执行成功，才确认消息投递，避免出现“消息发送失败，本地事务执行成功”或“消息发送成功，本地事务执行失败”的情况：

• 消息发送失败时，直接返回失败，不执行本地事务。

• 本地事务执行失败时，必须返回 Rollback，确保 Broker 删除半事务消息。

• 本地事务执行过程中，若出现异常，返回 UNKNOWN，依赖 Broker 回查机制兜底。

4.4 高并发场景优化
高并发场景下（如电商大促），大量事务消息会给 Broker 和生产者带来压力，优化建议：

• 生产者端：复用事务消息生产者实例，避免频繁创建；批量处理本地事务，减少数据库压力。

• Broker 端：增加 Broker 节点数量，优化线程池参数（sendMessageThreadPoolNums），提升消息处理能力；使用 SSD 磁盘，降低消息存储延迟。

• 消费者端：增加消费并行度（多实例、多线程），避免消息堆积；优化库存扣减逻辑，减少数据库锁竞争（如使用行锁）。

五、生产环境避坑指南（高频问题）

结合实际生产经验，总结以下4个高频坑，帮你规避分布式事务消息的一致性风险：

5.1 坑1：未实现消费幂等性，导致重复消费
误区：认为 RocketMQ 事务消息不会重复投递，未处理重复消费，导致库存重复扣减、订单重复创建等问题。

避坑方案：必须实现消费幂等性，通过“唯一标识去重”或“业务逻辑幂等”，确保重复消息消费结果一致；同时记录消费日志，便于排查重复消费问题。

5.2 坑2：本地事务未加事务注解，导致本地事务不生效
误区：生产者执行本地事务（如创建订单）时，未添加 @Transactional 注解，导致本地事务执行失败后无法回滚，出现“订单创建失败，消息仍被投递”的情况。

避坑方案：本地事务方法必须添加 @Transactional(rollbackFor = Exception.class) 注解，确保本地事务的原子性，执行失败时能正常回滚。

5.3 坑3：事务监听器未注入，导致事务消息发送失败

误区：未将 TransactionListener 实现类添加 @Component 注解，导致 RocketMQ 无法注入事务监听器，发送事务消息时抛出异常。

避坑方案：事务监听器必须添加 @Component 注解，确保被 Spring 容器管理，同时注入 OrderService、RocketMQTemplate 等依赖。

5.4 坑4：忽视死信队列，导致异常消息无人处理

误区：未监控死信队列，当消息因多次回查失败存入死信队列后，无人干预，导致数据不一致（如订单已创建，库存未扣减）。

避坑方案：通过 RocketMQ 控制台或监控工具（Prometheus + Grafana）监控死信队列，设置告警机制，一旦有异常消息，及时人工排查处理。

六、本篇核心总结及下一篇预告

• RocketMQ 分布式事务消息通过“半事务消息+事务确认+定时回查”机制，实现“本地业务操作与消息发送”的原子性，解决分布式系统的最终一致性问题，适合电商、支付等大多数分布式场景。

• 核心流程：发送半事务消息 → 执行本地事务 → 发送事务确认（Commit/Rollback） → 消费者消费消息；异常情况下，Broker 通过定时回查兜底，确保消息状态一致。

• 生产级关键：必须实现消费幂等性，避免重复消费；本地事务需添加事务注解，确保原子性；监控死信队列，及时处理异常消息；高并发场景下优化集群和代码性能。

• 与其他分布式事务方案相比，RocketMQ 事务消息开发成本低、性能好、无阻塞，依托 RocketMQ 集群的高可用性，无需额外搭建协调者，是分布式一致性的首选方案之一。

下一篇，我们将讲解 RocketMQ 高级特性——消息过滤、延迟消息与死信队列的深度应用，进一步拓展 RocketMQ 的业务适配能力，解决复杂场景下的消息处理难题。