消息中间件老炮重装上阵：Apache RocketMQ 5.x 架构师实战全景图

services:
  namesrv:
    image: apache/rocketmq:5.3.2
    container_name: rmq-namesrv
    ports:
      - "9876:9876"
    command: sh mqnamesrv
    environment:
      JAVA_OPT_EXT: "-Xms256m -Xmx256m"
    networks:
      - rocketmq

  broker:
    image: apache/rocketmq:5.3.2
    container_name: rmq-broker
    depends_on:
      - namesrv
    ports:
      - "10909:10909"
      - "10911:10911"
      - "10912:10912"
    environment:
      NAMESRV_ADDR: "rmq-namesrv:9876"
      JAVA_OPT_EXT: "-Xms512m -Xmx512m"
    command: sh mqbroker -n rmq-namesrv:9876 -c /home/rocketmq/broker.conf
    volumes:
      - ./broker.conf:/home/rocketmq/broker.conf:ro
    networks:
      - rocketmq

  proxy:
    image: apache/rocketmq:5.3.2
    container_name: rmq-proxy
    depends_on:
      - broker
      - namesrv
    ports:
      - "18080:8080"
      - "18081:8081"
    environment:
      NAMESRV_ADDR: "rmq-namesrv:9876"
      JAVA_OPT_EXT: "-Xms256m -Xmx256m"
    command: sh mqproxy
    restart: on-failure
    networks:
      - rocketmq

  dashboard:
    image: apacherocketmq/rocketmq-dashboard:latest
    container_name: rmq-dashboard
    depends_on:
      - namesrv
    ports:
      - "9999:8080"
    environment:
      JAVA_OPTS: "-Drocketmq.namesrv.addr=rmq-namesrv:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false"
    networks:
      - rocketmq

networks:
  rocketmq:
    driver: bridge

# 启动全部服务

docker compose up -d

# 查看服务状态

docker compose ps

# 查看 NameServer 日志

docker logs rmq-namesrv | grep "boot success"

# 验证 Broker 注册

docker exec rmq-broker sh mqadmin clusterList -n rmq-namesrv:9876

<parent>

    <groupId>org.springframework.boot</groupId>

    <artifactId>spring-boot-starter-parent</artifactId>

    <version>2.7.18</version>

</parent>

<dependencies>

    <dependency>

        <groupId>org.apache.rocketmq</groupId>

        <artifactId>rocketmq-spring-boot-starter</artifactId>

        <version>2.3.2</version>

    </dependency>

    <dependency>

        <groupId>org.springframework.boot</groupId>

        <artifactId>spring-boot-starter-web</artifactId>

    </dependency>

</dependencies>

rocketmq:

  name-server: 127.0.0.1:9876

  producer:

    group: demo-producer-group

    send-message-timeout: 3000

    retry-times-when-send-failed: 2

// 生产者：OrderController.java

@RestController

@RequestMapping("/order")

public class OrderController {

    @Autowired

 private RocketMQTemplate rocketMQTemplate;

 / POST /order/create?userId=1001&amount=299.00 /

    @PostMapping("/create")

 public String createOrder(String userId, BigDecimal amount) {

        OrderMessage msg = new OrderMessage(userId, amount, "ORDER\_" + System.currentTimeMillis());

 // 发送普通消息

        rocketMQTemplate.convertAndSend("ORDER\_NOTIFY\_TOPIC", msg);

 return "订单创建成功，通知已发送: " + msg.getOrderId();

    }

}

// 消费者：OrderNotifyConsumer.java

@Component

@RocketMQMessageListener(

    topic = "ORDER\_NOTIFY\_TOPIC",

    consumerGroup = "order-notify-consumer"

)

public class OrderNotifyConsumer implements RocketMQListener<OrderMessage> {

    @Override

 public void onMessage(OrderMessage message) {

 // 模拟发送短信

        System.out.printf("[短信通知] 用户 %s，您的订单 %s 已创建，金额 %.2f 元%n",

            message.getUserId(), message.getOrderId(), message.getAmount());

    }

}

curl -X POST "http://localhost:8080/order/create?userId=1001&amount=299.00"

# 控制台输出：[短信通知] 用户 1001，您的订单 ORDER\_xxx 已创建，金额 299.00 元

// 发送顺序消息（按 orderId 哈希分区）

@PostMapping("/order/status-flow")

public String orderStatusFlow(String orderId) {

    String[] statuses = {"CREATED", "PAID", "SHIPPED", "COMPLETED"};

 for (String status : statuses) {

 // 第三个参数是 hashKey，相同 orderId 进同一队列（有序）

        rocketMQTemplate.syncSendOrderly("ORDER\_STATUS\_TOPIC",

 new OrderStatusMsg(orderId, status), orderId);

    }

 return "顺序消息发送完成";

}

// 消费者：顺序消费

@Component

@RocketMQMessageListener(

    topic = "ORDER\_STATUS\_TOPIC",

    consumerGroup = "order-status-consumer",

    consumeMode = ConsumeMode.ORDERLY  // 关键：顺序消费模式

)

public class OrderStatusConsumer implements RocketMQListener<OrderStatusMsg> {

    @Override

 public void onMessage(OrderStatusMsg msg) {

        System.out.printf("[状态流转] 订单 %s → %s%n", msg.getOrderId(), msg.getStatus());

    }

}

// 下单时发送延迟消息（5s 用于测试，生产用 1800000ms）

@PostMapping("/order/with-timeout")

public String orderWithTimeout(String orderId) {

    Message<String> message = MessageBuilder

        .withPayload(orderId)

        .build();

 // delayMillis: 30分钟 = 1800000ms，测试用 10000ms

    rocketMQTemplate.syncSend("ORDER\_TIMEOUT\_TOPIC", message, 3000, 3);

 // RocketMQ 内置级别: 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

 return "订单 " + orderId + " 已创建，30分钟内请支付";

}

@Component

@RocketMQMessageListener(topic = "ORDER\_TIMEOUT\_TOPIC", consumerGroup = "order-timeout-consumer")

public class OrderTimeoutConsumer implements RocketMQListener<String> {

    @Override

 public void onMessage(String orderId) {

        System.out.printf("[超时取消] 订单 %s 超时未支付，执行自动取消%n", orderId);

 // 业务：查支付状态，未支付则取消订单

    }

}

// 发送事务消息

@PostMapping("/order/transaction")

public String transactionOrder(String productId, Integer quantity) {

    String orderId = "TXN\_" + System.currentTimeMillis();

    TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction(

 "ORDER\_TRANSACTION\_TOPIC",

        MessageBuilder.withPayload(new TxOrderMsg(orderId, productId, quantity)).build(),

        orderId // arg 传递给事务监听器

    );

 return "事务消息状态: " + result.getLocalTransactionState();

}

// 本地事务执行器

@Component

@RocketMQTransactionListener

public class OrderTransactionListener implements RocketMQLocalTransactionListener {

    @Override

 public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {

 try {

            String orderId = (String) arg;

 // 执行本地数据库操作（扣库存、创建订单）

 boolean success = inventoryService.deduct(orderId);

 return success ? COMMIT : ROLLBACK;

        } catch (Exception e) {

 return UNKNOWN; // 触发回查

        }

    }

    @Override

 public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {

 // Broker 回查本地事务状态（防止网络问题导致状态丢失）

        TxOrderMsg payload = JSON.parseObject(msg.getPayload().toString(), TxOrderMsg.class);

 boolean exists = orderService.existsOrder(payload.getOrderId());

 return exists ? COMMIT : ROLLBACK;

    }

}

// 发送带 Tag 的消息

public void sendWithTag(String userId, boolean isVip, String orderId) {

    String tag = isVip ? "VIP" : "NORMAL";

    rocketMQTemplate.convertAndSend("ORDER\_FILTER\_TOPIC:" + tag,

 new OrderMessage(userId, orderId));

}

// VIP 订单消费者（只消费 VIP Tag）

@Component

@RocketMQMessageListener(

    topic = "ORDER\_FILTER\_TOPIC",

    consumerGroup = "vip-order-consumer",

    selectorType = SelectorType.TAG,

    selectorExpression = "VIP" // 只消费 VIP 标签

)

public class VipOrderConsumer implements RocketMQListener<OrderMessage> {

    @Override

 public void onMessage(OrderMessage msg) {

        System.out.println("[VIP专属] 高优先级处理订单: " + msg.getOrderId());

    }

}

<!-- templates/mq-demo.html (Thymeleaf) -->

<!DOCTYPE html>

<html>

<head>

    <title>RocketMQ Demo Console</title>

    <style>

 body { font-family: 'Courier New', monospace; background: #1a1a2e; color: #e0e0e0; padding: 20px; }

 .card { background: #16213e; border: 1px solid #0f3460; border-radius: 8px; padding: 20px; margin: 10px 0; }

 button { background: #e94560; color: white; border: none; padding: 8px 20px; border-radius: 4px; cursor: pointer; }

 input, select { background: #0f3460; color: white; border: 1px solid #e94560; padding: 6px; border-radius: 4px; }

 .log { background: #000; padding: 10px; height: 200px; overflow-y: auto; font-size: 12px; }

    </style>

</head>

<body>

    <h2>🚀 RocketMQ 5.x Feature Demo</h2>

    <div class="card">

        <h3>消息类型测试</h3>

        <select id="msgType">

            <option value="normal">普通消息</option>

            <option value="delay">延迟消息（10s）</option>

            <option value="order">顺序消息</option>

            <option value="transaction">事务消息</option>

        </select>

        <input id="payload" placeholder="消息内容" value="Hello RocketMQ 5.x" />

        <button onclick="sendMsg()">发送</button>

    </div>

    <div class="card">

        <h3>消费日志</h3>

        <div class="log" id="log"></div>

    </div>

    <script>

 function sendMsg() {

 const type = document.getElementById('msgType').value;

 const payload = document.getElementById('payload').value;

 fetch(`/demo/send?type=${type}&payload=${encodeURIComponent(payload)}`)

                .then(r => r.text())

                .then(result => {

 document.getElementById('log').innerHTML +=

 `<div style="color:#4ecca3">[${new Date().toLocaleTimeString()}] ${result}</div>`;

                });

        }

 // 每 2s 刷新消费日志

 setInterval(() => {

 fetch('/demo/logs').then(r => r.json()).then(logs => {

 if (logs.length > 0) {

 const logDiv = document.getElementById('log');

 logDiv.innerHTML += logs.map(l =>

 `<div style="color:#ffd460">[消费] ${l}</div>`).join('');

                }

            });

        }, 2000);

    </script>

</body>

</html>

# 步骤1：每台机器拉取镜像

docker pull apache/rocketmq:5.3.2

# 步骤2：启动2个 NameServer（两台机器分别执行）

docker run -d --name rmq-ns \

  -p 9876:9876 \

  --restart always \

  apache/rocketmq:5.3.2 sh mqnamesrv

# 步骤3：broker.conf（Dledger 配置关键项）

cat > /data/rocketmq/broker-a.conf << EOF

brokerClusterName=RaftCluster

brokerName=broker-a

namesrvAddr=10.0.0.5:9876;10.0.0.6:9876

enableDLegerCommitLog=true

dLegerGroup=RaftNode00

dLegerPeers=n0-10.0.0.1:40911;n1-10.0.0.2:40911;n2-10.0.0.3:40911

dLegerSelfId=n0

storePathRootDir=/data/rocketmq/store

EOF

# 步骤4：启动 Broker（三副本 Raft 至少需要3节点）

docker run -d --name rmq-broker-a \

  -p 10911:10911 -p 40911:40911 \

  -v /data/rocketmq/broker-a.conf:/opt/rocketmq/conf/broker.conf \

  -v /data/rocketmq/store:/root/store \

  -e NAMESRV\_ADDR="10.0.0.5:9876;10.0.0.6:9876" \

  --restart always \

  apache/rocketmq:5.3.2 sh mqbroker -c /opt/rocketmq/conf/broker.conf

# 步骤5：验证集群状态

docker exec rmq-broker-a sh mqadmin clusterList -n 10.0.0.5:9876

# Kafka（KRaft 模式，无需 ZooKeeper）

docker run -d --name kafka \

  -p 9092:9092 \

  -e KAFKA\_CFG\_NODE\_ID=0 \

  -e KAFKA\_CFG\_PROCESS\_ROLES=controller,broker \

  -e KAFKA\_CFG\_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \

  -e KAFKA\_CFG\_CONTROLLER\_QUORUM\_VOTERS=0@localhost:9093 \

  bitnami/kafka:3.7

# RabbitMQ（含管理界面）

docker run -d --name rabbitmq \

  -p 5672:5672 -p 15672:15672 \

  -e RABBITMQ\_DEFAULT\_USER=admin \

  -e RABBITMQ\_DEFAULT\_PASS=admin123 \

  rabbitmq:3-management

# 访问：http://localhost:15672