在Spring Boot中使用RocketMQ事务消息进行消息的发送和接收的代码示例（极简可运行代码示例）

package com.example;

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

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

package com.example;

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

@Component
public class TxProducer {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    // 发送事务消息
    public void sendTxMsg(String msg) {
        // 构建事务消息（必须加事务ID）
        Message<String> txMessage = MessageBuilder
                .withPayload(msg)
                .setHeader(RocketMQHeaders.TRANSACTION_ID, UUID.randomUUID().toString())
                .build();

        // 发送半消息 + 绑定本地事务
        rocketMQTemplate.sendMessageInTransaction("tx-topic", txMessage, null);
        System.out.println("半消息发送成功：" + msg);
    }

    // 本地事务监听器（核心）
    @org.apache.rocketmq.spring.annotation.RocketMQTransactionListener(txProducerGroup = "tx-group")
    public class TxListener implements org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener {

        // 执行本地事务
        @Override
        public org.apache.rocketmq.spring.core.RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            String content = (String) msg.getPayload();
            try {
                // 模拟本地事务（成功则提交，失败则回滚）
                boolean success = true; // 改为false测试回滚
                if (success) {
                    System.out.println("本地事务执行成功，提交消息：" + content);
                    return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState.COMMIT;
                } else {
                    System.out.println("本地事务执行失败，回滚消息：" + content);
                    return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState.ROLLBACK;
                }
            } catch (Exception e) {
                return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState.UNKNOWN;
            }
        }

        // 事务回查（兜底）
        @Override
        public org.apache.rocketmq.spring.core.RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
            // 模拟回查：默认提交
            return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState.COMMIT;
        }
    }
}

package com.example;

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

@Component
@RocketMQMessageListener(topic = "tx-topic", consumerGroup = "tx-consumer-group")
public class TxConsumer implements RocketMQListener<String> {

    // 消费消息（仅接收提交的事务消息）
    @Override
    public void onMessage(String msg) {
        // 模拟消费逻辑（幂等校验可自行补充）
        System.out.println("消费事务消息：" + msg);
    }
}

package com.example;

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

@SpringBootTest
public class TxTest {
    @Autowired
    private TxProducer txProducer;

    @Test
    public void testTxMsg() throws InterruptedException {
        // 发送事务消息
        txProducer.sendTxMsg("Hello RocketMQ TX!");
        // 等待消费完成
        Thread.sleep(3000);
    }
}