Spring Boot 中 RocketMQ 事务消息完整配置指南

import com.alibaba.fastjson.JSON;
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 java.math.BigDecimal;
import java.util.UUID;

// 1. 业务DTO（以订单支付为例）
class OrderPayDTO {
    private String orderId;      // 订单ID
    private String userId;       // 用户ID
    private BigDecimal payAmount;// 支付金额

    // 构造器、getter/setter 省略
    public OrderPayDTO(String orderId, String userId, BigDecimal payAmount) {
        this.orderId = orderId;
        this.userId = userId;
        this.payAmount = payAmount;
    }
}

// 2. 事务消息生产者核心配置
@Component
public class RocketMQTxProducer {
    // 注入Spring Boot封装的RocketMQ模板
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送事务半消息（触发本地事务）
     */
    public void sendTxPayMessage(String orderId, String userId, BigDecimal payAmount) {
        // 1. 构建业务消息体
        OrderPayDTO dto = new OrderPayDTO(orderId, userId, payAmount);
        String msgBody = JSON.toJSONString(dto);

        // 2. 构建事务消息（必须设置唯一事务ID，避免回查冲突）
        Message<String> txMessage = MessageBuilder
                .withPayload(msgBody)
                .setHeader(RocketMQHeaders.TRANSACTION_ID, UUID.randomUUID().toString()) // 事务唯一标识
                .setHeader("orderId", orderId) // 自定义头（可选，便于回查）
                .build();

        // 3. 发送半消息 + 绑定本地事务监听器
        // 参数1：topic:tag（tag可选，用于消息过滤）
        // 参数2：事务消息体
        // 参数3：附加参数（可传递本地事务上下文，如userId）
        rocketMQTemplate.sendMessageInTransaction(
                "order_pay_topic:tx_pay_tag",
                txMessage,
                userId // 附加参数，会传递到本地事务方法
        );
    }

    /**
     * 本地事务监听器（核心：执行本地事务 + 事务回查）
     * 注解说明：txProducerGroup 必须与配置文件中生产者组一致！
     */
    @org.apache.rocketmq.spring.annotation.RocketMQTransactionListener(txProducerGroup = "tx-producer-group")
    public class OrderPayTxListener implements org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener {

        /**
         * 第一步：执行本地事务（核心业务逻辑，如扣减库存、更新订单状态）
         * @param msg 半消息内容
         * @param arg 发送半消息时传递的附加参数（此处为userId）
         * @return 事务状态：COMMIT(提交)/ROLLBACK(回滚)/UNKNOWN(未知，触发回查)
         */
        @Override
        public org.apache.rocketmq.spring.core.RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            try {
                // 解析消息体
                String msgBody = (String) msg.getPayload();
                OrderPayDTO dto = JSON.parseObject(msgBody, OrderPayDTO.class);
                String orderId = dto.getOrderId();
                String userId = (String) arg;

                // ========== 核心：执行本地事务（实际场景操作数据库）==========
                // 示例：1. 扣减用户余额  2. 更新订单为已支付
                boolean txSuccess = this.updateOrderStatus(orderId, "PAID") 
                        && this.deductUserBalance(userId, dto.getPayAmount());

                if (txSuccess) {
                    System.out.println("本地事务执行成功：订单" + orderId + "支付完成");
                    // 提交事务：Broker会投递消息给消费者
                    return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState.COMMIT;
                } else {
                    System.out.println("本地事务执行失败：订单" + orderId + "支付失败");
                    // 回滚事务：Broker会删除半消息，消费者不会收到
                    return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState.ROLLBACK;
                }
            } catch (Exception e) {
                System.err.println("本地事务执行异常：" + e.getMessage());
                // 返回UNKNOWN：Broker会定时回查（默认60秒一次）
                return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState.UNKNOWN;
            }
        }

        /**
         * 第二步：事务回查（解决生产者宕机/网络异常导致的状态丢失）
         * Broker会定时调用此方法，确认本地事务最终状态
         */
        @Override
        public org.apache.rocketmq.spring.core.RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
            try {
                // 解析消息体，获取订单ID
                String msgBody = (String) msg.getPayload();
                OrderPayDTO dto = JSON.parseObject(msgBody, OrderPayDTO.class);
                String orderId = dto.getOrderId();

                // ========== 核心：回查本地事务状态（查数据库/事务日志）==========
                String orderStatus = this.getOrderStatus(orderId);
                if ("PAID".equals(orderStatus)) {
                    // 本地事务已成功，提交消息
                    return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState.COMMIT;
                } else if ("UNPAID".equals(orderStatus)) {
                    // 本地事务未执行，回滚消息
                    return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState.ROLLBACK;
                } else {
                    // 状态未知，继续回查
                    return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState.UNKNOWN;
                }
            } catch (Exception e) {
                System.err.println("事务回查异常：" + e.getMessage());
                return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState.UNKNOWN;
            }
        }

        // 模拟：更新订单状态（实际场景操作数据库）
        private boolean updateOrderStatus(String orderId, String status) {
            // 业务逻辑：UPDATE t_order SET status = ? WHERE order_id = ?
            return true;
        }

        // 模拟：扣减用户余额（实际场景操作数据库）
        private boolean deductUserBalance(String userId, BigDecimal amount) {
            // 业务逻辑：UPDATE t_user SET balance = balance - ? WHERE user_id = ?
            return true;
        }

        // 模拟：查询订单状态（用于事务回查）
        private String getOrderStatus(String orderId) {
            // 业务逻辑：SELECT status FROM t_order WHERE order_id = ?
            return "PAID";
        }
    }
}

import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

@Component
// 监听事务消息的主题+标签，消费者组必须与配置文件一致
@RocketMQMessageListener(
        topic = "order_pay_topic",
        consumerGroup = "tx-consumer-group",
        selectorExpression = "tx_pay_tag", // 只消费tx_pay_tag标签的消息
        consumeMode = org.apache.rocketmq.common.protocol.heartbeat.ConsumeMode.CONCURRENTLY, // 并发消费
        messageModel = org.apache.rocketmq.common.protocol.heartbeat.MessageModel.CLUSTERING // 集群模式
)
public class OrderPayTxConsumer implements RocketMQListener<String> {

    /**
     * 消费事务消息（仅当生产者提交事务后，才会收到消息）
     * 注意：消费失败会自动重试，需保证幂等性！
     */
    @Override
    public void onMessage(String msg) {
        try {
            OrderPayDTO dto = JSON.parseObject(msg, OrderPayDTO.class);
            String orderId = dto.getOrderId();

            // ========== 幂等校验（核心！避免重复消费）==========
            if (this.checkConsumed(orderId)) {
                System.out.println("消息已消费：订单" + orderId + "，跳过处理");
                return;
            }

            // ========== 消费逻辑（如触发物流创建、发送短信通知）==========
            boolean consumeSuccess = this.createLogistics(orderId) && this.sendPaySms(dto.getUserId(), orderId);

            if (consumeSuccess) {
                System.out.println("消费成功：订单" + orderId + "，创建物流+发送短信完成");
                // 记录消费状态（用于幂等校验）
                this.recordConsumeStatus(orderId, "SUCCESS");
            } else {
                System.err.println("消费失败：订单" + orderId);
                // 抛出异常触发重试
                throw new RuntimeException("消费失败，重试");
            }
        } catch (Exception e) {
            System.err.println("消费异常：" + e.getMessage());
            throw new RuntimeException(e); // 触发重试（默认最多16次）
        }
    }

    // 模拟：幂等校验（查消费日志）
    private boolean checkConsumed(String orderId) {
        // SELECT COUNT(1) FROM t_consume_log WHERE order_id = ?
        return false;
    }

    // 模拟：记录消费状态
    private void recordConsumeStatus(String orderId, String status) {
        // INSERT INTO t_consume_log (order_id, status) VALUES (?, ?)
    }

    // 模拟：创建物流单
    private boolean createLogistics(String orderId) {
        return true;
    }

    // 模拟：发送支付成功短信
    private boolean sendPaySms(String userId, String orderId) {
        return true;
    }
}

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.math.BigDecimal;

@SpringBootTest
public class RocketMQTxTest {
    @Autowired
    private RocketMQTxProducer txProducer;

    @Test
    public void testTxMessage() throws InterruptedException {
        // 模拟订单支付：订单ID=ORDER_20250101，用户ID=USER_001，支付金额=100元
        txProducer.sendTxPayMessage("ORDER_20250101", "USER_001", new BigDecimal(100));
        
        // 等待事务执行和消费完成
        Thread.sleep(10000);
    }
}