Spring Boot中使用RocketMQ事务消息进行消息的发送和接收的代码示例

package com.example.rocketmqtxdemo.dto;

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;// 支付金额
}

package com.example.rocketmqtxdemo.producer;

import com.alibaba.fastjson.JSON;
import com.example.rocketmqtxdemo.dto.OrderPayDTO;
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 {

    @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. 发送半消息 + 绑定本地事务监听器
        rocketMQTemplate.sendMessageInTransaction(
                "order_pay_topic:tx_tag",  // 主题+标签
                txMessage,                 // 消息体
                userId                     // 附加参数（传递到本地事务方法）
        );
        System.out.println("【生产者】半消息发送成功，订单ID：" + orderId);
    }

    /**
     * 本地事务监听器（核心：执行本地事务 + 事务回查）
     */
    @org.apache.rocketmq.spring.annotation.RocketMQTransactionListener(txProducerGroup = "tx-producer-group")
    public class OrderPayTxListener implements org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener {

        /**
         * 执行本地事务（核心业务逻辑：扣减余额、更新订单状态）
         */
        @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;

                // ========== 模拟本地事务：扣减用户余额 + 更新订单状态 ==========
                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回查
                return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState.UNKNOWN;
            }
        }

        /**
         * 事务回查（解决：本地事务成功但COMMIT指令未送达Broker）
         */
        @Override
        public org.apache.rocketmq.spring.core.RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
            try {
                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("【生产者】事务回查成功，提交消息，订单ID：" + orderId);
                    return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState.COMMIT;
                } else {
                    System.out.println("【生产者】事务回查失败，回滚消息，订单ID：" + orderId);
                    return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState.ROLLBACK;
                }
            } 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";
        }
    }
}

package com.example.rocketmqtxdemo.consumer;

import com.alibaba.fastjson.JSON;
import com.example.rocketmqtxdemo.dto.OrderPayDTO;
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", // 与配置文件消费者组一致
        selectorExpression = "tx_tag"     // 过滤标签
)
public class TxMessageConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String msg) {
        try {
            // 1. 解析消息体
            OrderPayDTO dto = JSON.parseObject(msg, OrderPayDTO.class);
            String orderId = dto.getOrderId();
            System.out.println("【消费者】收到事务消息，订单ID：" + orderId);

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

            // 3. 模拟消费逻辑：创建物流单 + 发送短信
            boolean createLogistics = createLogistics(orderId);
            boolean sendSms = sendPaySuccessSms(dto.getUserId(), orderId);

            if (createLogistics && sendSms) {
                // 4. 记录消费状态（落地到数据库）
                recordConsumeStatus(orderId);
                System.out.println("【消费者】消息消费成功，订单ID：" + orderId);
            } else {
                System.err.println("【消费者】消息消费失败，订单ID：" + orderId);
                throw new RuntimeException("消费失败，触发重试");
            }
        } catch (Exception e) {
            System.err.println("【消费者】消费异常：" + e.getMessage());
            throw new RuntimeException(e); // 抛出异常触发重试
        }
    }

    // ------------------- 模拟消费逻辑 -------------------
    // 幂等校验：查询是否已消费
    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) {
        return true;
    }

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

package com.example.rocketmqtxdemo;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class RocketmqTxDemoApplication {
    public static void main(String[] args) {
        SpringApplication.run(RocketmqTxDemoApplication.class, args);
    }
}

package com.example.rocketmqtxdemo;

import com.example.rocketmqtxdemo.producer.TxMessageProducer;
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 {
        // 模拟发送订单支付事务消息
        txProducer.sendTxPayMessage("ORDER_20250101", "USER_001", new BigDecimal(100));
        
        // 等待事务执行和消费完成（测试用，生产环境无需休眠）
        Thread.sleep(10000);
    }
}