Spring Boot 中 RocketMQ 事务消息的发送与接收（完整实操）

import lombok.AllArgsConstructor;
import lombok.Data;
import java.math.BigDecimal;

// 订单支付消息体
@Data
@AllArgsConstructor
public class OrderPayDTO {
    private String orderId;      // 订单ID（业务唯一标识）
    private String userId;       // 用户ID
    private BigDecimal payAmount;// 支付金额
}

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.Service;
import java.util.UUID;

@Service
public class TxMessageProducer {
    // 注入 Spring Boot 封装的 RocketMQ 模板
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 步骤1：发送事务半消息（触发本地事务）
     * @param orderId 订单ID
     * @param userId 用户ID
     * @param payAmount 支付金额
     */
    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)
                // 事务唯一标识（RocketMQ 用于回查）
                .setHeader(RocketMQHeaders.TRANSACTION_ID, UUID.randomUUID().toString())
                // 自定义头（便于回查时快速获取关键信息）
                .setHeader("orderId", orderId)
                .build();

        // 3. 发送半消息 + 绑定本地事务监听器
        // 参数说明：
        // - "order_pay_topic:tx_tag"：主题+标签（标签用于消息过滤）
        // - txMessage：事务消息体
        // - userId：附加参数（传递到本地事务方法）
        rocketMQTemplate.sendMessageInTransaction(
                "order_pay_topic:tx_tag",
                txMessage,
                userId
        );
        System.out.println("半消息发送成功，订单ID：" + orderId);
    }

    /**
     * 步骤2：本地事务监听器（核心：执行本地事务 + 事务回查）
     * 注解说明：txProducerGroup 必须与 application.yml 中生产者组一致！
     */
    @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 deductSuccess = deductUserBalance(userId, dto.getPayAmount());
                boolean updateOrderSuccess = updateOrderStatus(orderId, "PAID");

                if (deductSuccess && updateOrderSuccess) {
                    System.out.println("本地事务执行成功，订单ID：" + orderId);
                    // 提交事务：Broker 会将半消息转为可消费状态
                    return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState.COMMIT;
                } else {
                    System.out.println("本地事务执行失败，订单ID：" + 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;
            }
        }

        /**
         * 事务回查（解决：本地事务成功，但 COMMIT 指令未送达 Broker 的场景）
         * 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 = getOrderStatusFromDB(orderId);
                if ("PAID".equals(orderStatus)) {
                    System.out.println("事务回查成功，订单已支付，提交消息：" + orderId);
                    return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState.COMMIT;
                } else if ("UNPAID".equals(orderStatus)) {
                    System.out.println("事务回查失败，订单未支付，回滚消息：" + orderId);
                    return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState.ROLLBACK;
                } else {
                    // 状态未知，继续回查（最多回查15次，可配置）
                    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 deductUserBalance(String userId, BigDecimal amount) {
            // 实际逻辑：UPDATE t_user SET balance = balance - ? WHERE user_id = ?
            return true; // 模拟成功，可改为 false 测试回滚
        }

        // 模拟更新订单状态
        private boolean updateOrderStatus(String orderId, String status) {
            // 实际逻辑：UPDATE t_order SET status = ? WHERE order_id = ?
            return true;
        }

        // 模拟查询订单状态（用于事务回查）
        private String getOrderStatusFromDB(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.Service;

@Service
// 监听配置：主题+消费者组+标签过滤
@RocketMQMessageListener(
        topic = "order_pay_topic",        // 与生产者发送的主题一致
        consumerGroup = "tx-consumer-group", // 与 application.yml 中消费者组一致
        selectorExpression = "tx_tag",    // 只消费 tx_tag 标签的消息
        consumeMode = org.apache.rocketmq.common.protocol.heartbeat.ConsumeMode.CONCURRENTLY, // 并发消费
        messageModel = org.apache.rocketmq.common.protocol.heartbeat.MessageModel.CLUSTERING // 集群模式
)
public class TxMessageConsumer implements RocketMQListener<String> {

    /**
     * 消费事务消息（仅当生产者提交事务后，才会触发此方法）
     * @param msg 消息体（与生产者发送的内容一致）
     */
    @Override
    public void onMessage(String msg) {
        try {
            // 1. 解析消息体
            OrderPayDTO dto = JSON.parseObject(msg, OrderPayDTO.class);
            String orderId = dto.getOrderId();
            System.out.println("收到事务消息，开始消费：" + orderId);

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

            // 3. 执行业务逻辑（如创建物流、发送短信）
            boolean createLogistics = createLogistics(orderId);
            boolean sendSms = sendPaySuccessSms(dto.getUserId(), orderId);

            if (createLogistics && sendSms) {
                // 4. 记录消费状态（用于幂等校验）
                recordConsumeStatus(orderId);
                System.out.println("消息消费成功：" + orderId);
            } else {
                System.err.println("消息消费失败：" + orderId);
                // 抛出异常触发重试（默认重试3次，可配置）
                throw new RuntimeException("消费失败，触发重试");
            }
        } catch (Exception e) {
            System.err.println("消费异常：" + e.getMessage());
            throw new RuntimeException(e); // 必须抛出异常，否则 RocketMQ 认为消费成功
        }
    }

    // ------------------- 模拟消费逻辑 -------------------
    // 幂等校验：查询是否已消费过该订单
    private boolean checkConsumed(String orderId) {
        // 实际逻辑：SELECT COUNT(1) FROM t_consume_log WHERE order_id = ?
        return false; // 模拟未消费
    }

    // 记录消费状态（落地到数据库）
    private void recordConsumeStatus(String orderId) {
        // 实际逻辑：INSERT INTO t_consume_log (order_id, consume_time) VALUES (?, NOW())
    }

    // 模拟创建物流单
    private boolean createLogistics(String orderId) {
        // 实际逻辑：INSERT INTO t_logistics (order_id, status) VALUES (?, 'CREATE')
        return true;
    }

    // 模拟发送支付成功短信
    private boolean sendPaySuccessSms(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 TxMessageProducer txProducer;

    @Test
    public void testTxMessageSendAndConsume() throws InterruptedException {
        // 模拟订单支付：订单ID=ORDER_20250101，用户ID=USER_001，支付金额=100元
        txProducer.sendTxPayMessage("ORDER_20250101", "USER_001", new BigDecimal(100));
        
        // 等待事务执行和消费完成（实际生产无需休眠，此处为测试）
        Thread.sleep(10000);
    }
}