RocketMQ 4.X 消息队列：从入门到实战的完全指南

┌─────────────────┐
│   NameServer    │ ← 无状态路由中心（去中心化）
│  (集群部署)      │
└────────┬────────┘
         │ 路由发现
    ┌────┴────┐
    ▼         ▼         ▼
┌────────┐ ┌────────┐ ┌────────┐
│Producer│ │ Broker │ │Consumer│
│(生产者)│→│(主从集群)│←│(消费者)│
└────────┘ └────────┘ └────────┘

xml<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.3</version>
</dependency>

yamlrocketmq:
  name-server: ${ROCKETMQ_NAMESRV:127.0.0.1:9876}
mq:
  delayQueue:
    topic: delay_queue_topic
    tag: delay_queue_tag
    group: delay_queue_group

java@Component
public class PayProducer {
    private final DefaultMQProducer producer;

    public PayProducer(@Value("${rocketmq.name-server}") String namesrv) {
        this.producer = new DefaultMQProducer("pay_producer_group");
        this.producer.setNamesrvAddr(namesrv);
        this.producer.setCompressMsgBodyOverHowmuch(1024 * 1024); // 1MB以上压缩
        this.producer.setRetryTimesWhenSendFailed(5); // 重试5次
        this.producer.start();
    }

    public SendResult send(String topic, String tag, String body) throws Exception {
        Message msg = new Message(topic, tag, body.getBytes(RemotingHelper.DEFAULT_CHARSET));
        return producer.send(msg);
    }
}

java@Component
public class PayConsumer {
    
    @PostConstruct
    public void init() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("pay_consumer_group");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.setConsumeThreadMin(20);
        consumer.setConsumeThreadMax(64);
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.subscribe("pay_topic", "*");
        
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (Message msg : msgs) {
                System.out.printf("收到消息: %s%n", new String(msg.getBody()));
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        
        consumer.start();
    }
}

阶段1: 订单服务发送"半事务消息" → 消费者不可见
阶段2a: 本地事务成功（插入订单）→ 发送"确认提交" → 库存服务扣减 ✅
阶段2b: 本地事务失败 → 发送"回滚" → 消息丢弃 ❌

bash# 1. 启动 NameServer
nohup sh bin/mqnamesrv &

# 2. 启动 Broker（自动创建Topic）
nohup sh bin/mqbroker -n127.0.0.1:9876 autoCreateTopicEnable=true &

bash# 查看所有Topic
sh bin/mqadmin topicList -n127.0.0.1:9876

# 查看Topic路由
sh bin/mqadmin topicRoute -n127.0.0.1:9876 -t your_topic