消息队列：Kafka / RocketMQ / RabbitMQ 三件套选型与生产部署

1
2
3
4
5
6

partition-0/
├── 00000000000000000000.log      # 消息数据
├── 00000000000000000000.index     # 位移索引
├── 00000000000000000000.timeindex # 时间戳索引
├── 00000000000123456789.log
└── ...

1
2
3
4

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

spring:
  kafka:
    bootstrap-servers: kafka-cluster:9092
    producer:
      acks: all
      retries: 3
      properties:
        enable.idempotence: true           # 开启幂等
        max.in.flight.requests.per.connection: 5
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      group-id: order-consumer
      auto-offset-reset: earliest
      enable-auto-commit: false            # 关闭自动提交，手动 ack
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
    listener:
      ack-mode: manual_immediate

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

@Service
public class OrderEventProducer {
    @Autowired
    private KafkaTemplate<String, OrderEvent> kafkaTemplate;

    public void sendOrderCreated(OrderEvent event) {
        // key=orderId 保证同一订单消息落到同一 Partition
        CompletableFuture<SendResult<String, OrderEvent>> future =
            kafkaTemplate.send("order-topic", event.getOrderId(), event);
        future.whenComplete((result, ex) -> {
            if (ex == null) {
                log.info("消息发送成功: orderId={}, offset={}",
                    event.getOrderId(), result.getRecordMetadata().offset());
            } else {
                log.error("消息发送失败", ex);
                // 业务侧重试或告警
            }
        });
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

@Component
public class OrderEventConsumer {
    @KafkaListener(topics = "order-topic", groupId = "order-consumer")
    public void onMessage(
        @Payload OrderEvent event,
        @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
        @Header(KafkaHeaders.OFFSET) long offset,
        Acknowledgment ack
    ) {
        try {
            // 业务处理
            orderService.handle(event);
            // 手动 ack
            ack.acknowledge();
        } catch (BusinessException e) {
            // 业务异常：nack 不重试（避免死循环）
            log.warn("业务异常，跳过: {}", e.getMessage());
            ack.acknowledge();
        } catch (Exception e) {
            // 系统异常：抛出让 Spring Kafka 重试
            throw e;
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

@Bean
public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> pf) {
    KafkaTemplate<String, String> template = new KafkaTemplate<>(pf);
    template.setTransactionIdPrefix("order-tx-");
    return template;
}

@Service
public class OrderTxService {
    @Transactional("kafkaTransactionManager")
    public void createOrderWithEvent(Order order) {
        // 1. 写订单库
        orderRepository.save(order);
        // 2. 发 Kafka 消息（事务保证原子性）
        kafkaTemplate.send("order-topic", order.getId(), toEvent(order));
        // 任一失败，整体回滚
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

version: "3.8"

# ============= Controller 集群（KRaft 模式） =============
services:
  kafka-controller-1:
    image: bitnami/kafka:3.7
    container_name: kafka-controller-1
    ports:
      - "9093:9093"
    environment:
      # KRaft 模式关键配置
      KAFKA_CFG_NODE_ID: 1
      KAFKA_CFG_PROCESS_ROLES: "controller,broker"  # 同节点既做 controller 也做 broker（小集群）
      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: "1@kafka-controller-1:9093,2@kafka-controller-2:9093,3@kafka-controller-3:9093"
      KAFKA_CFG_LISTENERS: "PLAINTEXT://:9092,CONTROLLER://:9093"
      KAFKA_CFG_ADVERTISED_LISTENERS: "PLAINTEXT://kafka-controller-1:9092"
      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: "PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT"
      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
      KAFKA_CFG_INTER_BROKER_LISTENER_NAME: "PLAINTEXT"
      # 持久化
      KAFKA_CFG_LOG_DIRS: "/bitnami/kafka/data"
      # JMX 导出给 Prometheus
      KAFKA_JMX_PORT: 9095
      KAFKA_JMX_HOSTNAME: localhost
      JMX_PROMETHEUS_PORT: 9095
    volumes:
      - kafka1-data:/bitnami/kafka/data
    networks:
      - kafka-net

  kafka-controller-2:
    image: bitnami/kafka:3.7
    container_name: kafka-controller-2
    ports:
      - "9094:9093"
    environment:
      KAFKA_CFG_NODE_ID: 2
      KAFKA_CFG_PROCESS_ROLES: "controller,broker"
      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: "1@kafka-controller-1:9093,2@kafka-controller-2:9093,3@kafka-controller-3:9093"
      KAFKA_CFG_LISTENERS: "PLAINTEXT://:9092,CONTROLLER://:9093"
      KAFKA_CFG_ADVERTISED_LISTENERS: "PLAINTEXT://kafka-controller-2:9092"
      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: "PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT"
      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
      KAFKA_CFG_INTER_BROKER_LISTENER_NAME: "PLAINTEXT"
      KAFKA_CFG_LOG_DIRS: "/bitnami/kafka/data"
      KAFKA_JMX_PORT: 9095
    volumes:
      - kafka2-data:/bitnami/kafka/data
    networks:
      - kafka-net

  kafka-controller-3:
    image: bitnami/kafka:3.7
    container_name: kafka-controller-3
    ports:
      - "9095:9093"
    environment:
      KAFKA_CFG_NODE_ID: 3
      KAFKA_CFG_PROCESS_ROLES: "controller,broker"
      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: "1@kafka-controller-1:9093,2@kafka-controller-2:9093,3@kafka-controller-3:9093"
      KAFKA_CFG_LISTENERS: "PLAINTEXT://:9092,CONTROLLER://:9093"
      KAFKA_CFG_ADVERTISED_LISTENERS: "PLAINTEXT://kafka-controller-3:9092"
      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: "PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT"
      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
      KAFKA_CFG_INTER_BROKER_LISTENER_NAME: "PLAINTEXT"
      KAFKA_CFG_LOG_DIRS: "/bitnami/kafka/data"
      KAFKA_JMX_PORT: 9095
    volumes:
      - kafka3-data:/bitnami/kafka/data
    networks:
      - kafka-net

  # ============= Kafka UI（管理界面） =============
  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    container_name: kafka-ui
    ports:
      - "8080:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: "kafka-cluster"
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: "kafka-controller-1:9092,kafka-controller-2:9092,kafka-controller-3:9092"
    depends_on:
      - kafka-controller-1
      - kafka-controller-2
      - kafka-controller-3
    networks:
      - kafka-net

volumes:
  kafka1-data:
  kafka2-data:
  kafka3-data:

networks:
  kafka-net:
    driver: bridge

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# 1. 启动集群
docker-compose up -d

# 2. 验证 Controller 集群状态
docker exec kafka-controller-1 kafka-metadata-quorum --bootstrap-server kafka-controller-1:9092 describe --status
# 应显示 3 个 nodeId 都在线

# 3. 创建 Topic（3 分区 3 副本）
docker exec kafka-controller-1 kafka-topics --create \
  --bootstrap-server kafka-controller-1:9092 \
  --topic order-topic \
  --partitions 3 \
  --replication-factor 3

# 4. 验证 Topic
docker exec kafka-controller-1 kafka-topics --list --bootstrap-server kafka-controller-1:9092
# 应输出：order-topic

# 5. 访问 UI
# 浏览器打开 http://localhost:8080

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# 演练 1：杀 Broker 节点
docker stop kafka-controller-2
# 观察：剩余 2 节点仍可服务（replication-factor=3，2 副本够用）
# 30 秒内 kafka-ui 应显示 Broker 2 offline

# 演练 2：杀 Controller 节点（半数以上存活）
docker stop kafka-controller-2
docker stop kafka-controller-3
# 观察：只剩 1 节点 Controller，集群**仍可读不可写**（写入需要 Controller 仲裁）
# 这是为什么 Controller 必须 3 节点（容忍 1 节点故障）

# 演练 3：网络分区模拟
docker network disconnect kafka-net kafka-controller-2
# 观察：Controller 2 被网络隔离，集群 30 秒内完成 Controller 重选

1

业务系统 → Kafka（数据采集层）→ Flink（实时计算层）→ Kafka / MySQL / ES（结果层）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

store/
├── CommitLog/         # 所有消息顺序写（1 个文件 1GB，默认）
│   ├── 00000000000000000000
│   └── 00000000001073741824
├── ConsumeQueue/      # 消费队列（每个 MessageQueue 一个）
│   └── order-topic/
│       └── 0/         # Queue 0 的消费索引
│           ├── 00000000000000000000
│           └── ...
├── Index/             # 按 key / 时间戳 索引
│   └── 202606091230
└── config/            # Broker 配置

1
2
3
4
5
6
7

1. Producer 发送 prepared 消息（half message）→ RocketMQ
2. RocketMQ 标记消息为"prepared"（消费者不可见）
3. Producer 执行本地事务（如创建订单）
4. Producer 根据本地事务结果发 commit 或 rollback
   - commit → RocketMQ 让消息对消费者可见
   - rollback → RocketMQ 删除 prepared 消息
5. 如果 Producer 阶段 4 失败，RocketMQ **定期回查** Producer 的本地事务状态

1
2
3
4
5

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

rocketmq:
  name-server: rocketmq-namesrv-1:9876,rocketmq-namesrv-2:9876
  producer:
    group: order-producer
    send-message-timeout: 10000
    retry-times-when-send-failed: 3
    retry-times-when-send-async-failed: 3
    max-message-size: 4194304      # 4MB
    compress-message-body-threshold: 4096
  consumer:
    group: order-consumer
    enable-message-trace: false
    pull-batch-size: 32

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

@Service
public class OrderTxProducer {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void createOrderWithEvent(Order order) {
        // 发送事务消息
        rocketMQTemplate.sendMessageInTransaction(
            "order-tx-topic",          // Topic
            MessageBuilder.withPayload(order).build(),
            order                       // arg 传给本地事务监听器
        );
    }

    @RocketMQTransactionListener(txProducerGroup = "order-tx-producer")
    public class OrderTransactionListener implements RocketMQLocalTransactionListener {
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            try {
                // 1. 本地事务：创建订单（写 DB）
                Order order = (Order) arg;
                orderRepository.save(order);
                // 2. 成功 → 提交消息
                return RocketMQLocalTransactionState.COMMIT;
            } catch (Exception e) {
                // 失败 → 回滚消息
                return RocketMQLocalTransactionState.ROLLBACK;
            }
        }

        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
            // 回查：RocketMQ 询问未确认消息的本地事务状态
            String orderId = msg.getKeys();
            boolean exists = orderRepository.existsById(orderId);
            return exists
                ? RocketMQLocalTransactionState.COMMIT
                : RocketMQLocalTransactionState.ROLLBACK;
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

@Component
@RocketMQMessageListener(
    topic = "order-topic",
    consumerGroup = "order-consumer",
    consumeMode = ConsumeMode.ORDERLY   // 顺序消费模式
)
public class OrderSeqConsumer implements RocketMQListener<OrderEvent> {
    @Override
    public void onMessage(OrderEvent event) {
        // 业务处理
        orderService.handle(event);
    }
}

// 生产者侧用 MessageQueueSelector 保证同 orderId 路由到同队列
public void sendOrderEvent(OrderEvent event) {
    rocketMQTemplate.syncSendOrderly(
        "order-topic",
        MessageBuilder.withPayload(event).build(),
        event.getOrderId(),  // hashKey
        3000                 // 超时
    );
}

1
2
3
4
5
6
7
8

@RocketMQMessageListener(
    topic = "config-topic",
    consumerGroup = "config-broadcast",
    messageModel = MessageModel.BROADCASTING   // 广播模式
)
public class ConfigConsumer implements RocketMQListener<ConfigEvent> {
    // 每个 Consumer 收到所有消息（适合配置刷新场景）
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102

version: "3.8"

services:
  # ============= NameServer 集群 =============
  rocketmq-namesrv-1:
    image: apache/rocketmq:5.3.0
    container_name: rocketmq-namesrv-1
    command: sh mqnamesrv
    ports:
      - "9876:9876"
    volumes:
      - namesrv1-logs:/home/rocketmq/logs
    networks:
      - rocketmq-net

  rocketmq-namesrv-2:
    image: apache/rocketmq:5.3.0
    container_name: rocketmq-namesrv-2
    command: sh mqnamesrv
    ports:
      - "9877:9876"
    volumes:
      - namesrv2-logs:/home/rocketmq/logs
    networks:
      - rocketmq-net

  # ============= Controller 集群（5.x 负责 Dledger 选主） =============
  rocketmq-controller:
    image: apache/rocketmq:5.3.0
    container_name: rocketmq-controller
    command: sh mqcontroller
    ports:
      - "8080:8080"
    environment:
      JAVA_OPT_EXT: "-Xms512m -Xmx512m"
    volumes:
      - controller-logs:/home/rocketmq/logs
    networks:
      - rocketmq-net

  # ============= Broker 节点 1（Dledger 副本组 1） =============
  rocketmq-broker-1:
    image: apache/rocketmq:5.3.0
    container_name: rocketmq-broker-1
    command: sh mqbroker
    ports:
      - "10911:10911"   # Broker 主端口
      - "10912:10912"   # Broker HA 端口
    environment:
      JAVA_OPT_EXT: "-Xms1g -Xmx1g -Xmn512m"
      NAMESRV_ADDR: "rocketmq-namesrv-1:9876;rocketmq-namesrv-2:9876"
    volumes:
      - broker1-store:/home/rocketmq/store
      - broker1-logs:/home/rocketmq/logs
      - ./broker-1.conf:/home/rocketmq/rocketmq-5.3.0/conf/broker.conf
    depends_on:
      - rocketmq-namesrv-1
      - rocketmq-namesrv-2
      - rocketmq-controller
    networks:
      - rocketmq-net

  # ============= Broker 节点 2、3、4（省略，依同样模式） =============
  # 略：rocketmq-broker-2、3、4 同样配置
  # 重要：每个 broker.conf 里需要：
  #   brokerClusterName=DefaultCluster
  #   brokerName=broker-a
  #   brokerId=0 (master) / 1 (slave)
  #   brokerRole=DLEDGER (5.x 必须)
  #   listenPort=10911
  #   namesrvAddr=rocketmq-namesrv-1:9876;rocketmq-namesrv-2:9876
  #   storePathRootDir=/home/rocketmq/store
  #   dledgerGroup=broker-a-dledger  # 同副本组同名
  #   dledgerPeers=n0;rocketmq-broker-1:40911;n1;rocketmq-broker-2:40911;n2;rocketmq-broker-3:40911
  #   dledgerSelfId=n0  # 当前节点 ID
  #   enableDLegerCommitLog=true

  # ============= RocketMQ Console =============
  rocketmq-console:
    image: apacherocketmq/rocketmq-console-ng:latest
    container_name: rocketmq-console
    ports:
      - "8081:8080"
    environment:
      JAVA_OPTS: "-Drocketmq.namesrv.addr=rocketmq-namesrv-1:9876;rocketmq-namesrv-2:9876"
    depends_on:
      - rocketmq-namesrv-1
      - rocketmq-namesrv-2
    networks:
      - rocketmq-net

volumes:
  namesrv1-logs:
  namesrv2-logs:
  controller-logs:
  broker1-store:
  broker1-logs:
  # 其他 broker 同

networks:
  rocketmq-net:
    driver: bridge

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

brokerClusterName=DefaultCluster
brokerName=broker-a
brokerId=0
brokerRole=DLEDGER
listenPort=10911
namesrvAddr=rocketmq-namesrv-1:9876;rocketmq-namesrv-2:9876
storePathRootDir=/home/rocketmq/store
dledgerGroup=broker-a-dledger
dledgerPeers=n0;rocketmq-broker-1:40911;n1;rocketmq-broker-2:40911;n2;rocketmq-broker-3:40911
dledgerSelfId=n0
enableDLegerCommitLog=true
flushCommitLogLeastPages=4
flushCommitLogThoroughInterval=10000

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

docker-compose up -d
# 等待 30 秒（启动顺序：NameServer → Controller → Broker）

# 1. 验证 NameServer
docker exec rocketmq-namesrv-1 sh mqadmin clusterList -n rocketmq-namesrv-1:9876
# 应看到所有 Broker 节点

# 2. 创建 Topic
docker exec rocketmq-broker-1 sh mqadmin updateTopic \
  -n rocketmq-namesrv-1:9876 \
  -t order-topic \
  -c DefaultCluster \
  -b rocketmq-broker-1:10911 \
  -r 4 -w 4   # 4 读 4 写队列

# 3. 发送测试消息
docker exec rocketmq-broker-1 sh tools.sh org.apache.rocketmq.example.quickstart.Producer

# 4. 访问 Console
# 浏览器打开 http://localhost:8081

1
2
3
4
5
6
7
8
9

# 演练 1：杀 Broker Master
docker stop rocketmq-broker-1
# 观察：rocketmq-broker-2 自动升级为新 Master（Dledger Raft 选主，30 秒内）
# 验证：rocketmq-console 看 broker-2 role=MASTER

# 演练 2：杀 NameServer
docker stop rocketmq-namesrv-1
# 观察：剩 1 个 NameServer，**生产消费仍可继续**（NameServer 无状态，路由信息已缓存）
# 启动新的 NameServer 自动加入集群

1
2
3
4
5
6
7

# 正常状态
INFO  Dledger - dledger[broker-a-dledger] elected new leader: n0
INFO  BrokerController - broker-a 切换为新 Master

# 异常状态（split brain）
WARN  Dledger - 发现两个节点都认为自己是 Leader
ERROR BrokerController - Dledger 选主冲突，需要人工介入

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# 查看集群状态
mqadmin clusterList -n namesrv:9876

# 查看 Topic 列表
mqadmin topicList -n namesrv:9876

# 查看消费进度
mqadmin consumerProgress -n namesrv:9876 -g consumer-group

# 重置消费位点
mqadmin resetOffsetByTime -n namesrv:9876 -t topic -g group -s timestamp

# 查看 Broker 状态
mqadmin brokerStatus -n namesrv:9876 -b broker:10911

1
2
3
4

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

spring:
  rabbitmq:
    host: rabbitmq-cluster
    port: 5672
    username: admin
    password: ${RABBITMQ_PASSWORD}
    virtual-host: /order
    publisher-confirm-type: correlated   # 开启 Publisher Confirm
    publisher-returns: true              # 开启消息路由失败回调
    listener:
      simple:
        acknowledge-mode: manual         # 手动 ack
        prefetch: 32                     # 消费者预取
        retry:
          enabled: true
          max-attempts: 3
          initial-interval: 2s

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

@Configuration
public class RabbitConfig {
    @Bean
    public TopicExchange orderExchange() {
        return ExchangeBuilder
            .topicExchange("order-events")
            .durable(true)
            .build();
    }

    @Bean
    public Queue orderCreatedQueue() {
        return QueueBuilder.durable("order-created-quorum")
            .quorum()                        // ⭐ Quorum Queue
            .deadLetterExchange("order-dlx")  // 死信交换器
            .deadLetterRoutingKey("dlx")
            .ttl(86400000)                    # 消息 TTL 24 小时
            .build();
    }

    @Bean
    public Binding orderCreatedBinding(Queue orderCreatedQueue, TopicExchange orderExchange) {
        return BindingBuilder
            .bind(orderCreatedQueue)
            .to(orderExchange)
            .with("order.created");           // routing key 通配
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

@Service
public class OrderEventPublisher {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void publishOrderCreated(OrderEvent event) {
        // 设置 ConfirmCallback
        rabbitTemplate.setConfirmCallback((correlation, ack, cause) -> {
            if (!ack) {
                log.error("消息未确认: {}", cause);
                // 重试或告警
            }
        });

        // 设置 ReturnsCallback（消息路由失败回调）
        rabbitTemplate.setReturnsCallback(returned ->
            log.error("消息路由失败: exchange={}, routingKey={}",
                returned.getExchange(), returned.getRoutingKey())
        );

        rabbitTemplate.convertAndSend(
            "order-events",
            "order.created",
            event,
            message -> {
                message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
                return message;
            }
        );
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

@Component
public class OrderCreatedConsumer {
    @RabbitListener(queues = "order-created-quorum")
    public void onMessage(
        @Payload OrderEvent event,
        Channel channel,
        @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag
    ) {
        try {
            // 业务处理
            inventoryService.deduct(event);
            // 手动 ack
            channel.basicAck(deliveryTag, false);
        } catch (BusinessException e) {
            // 业务异常：nack 不重试（走死信队列）
            channel.basicNack(deliveryTag, false, false);
        } catch (Exception e) {
            // 系统异常：nack 重试
            channel.basicNack(deliveryTag, false, true);
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

// 1. 启用插件：rabbitmq-plugins enable rabbitmq_delayed_message_exchange
// 2. 声明延迟 Exchange
@Bean
public CustomExchange delayedExchange() {
    Map<String, Object> args = new HashMap<>();
    args.put("x-delayed-type", "direct");
    return new CustomExchange("order-delayed", "x-delayed-message", true, false, args);
}

// 3. 发送延迟消息（30 分钟）
rabbitTemplate.convertAndSend(
    "order-delayed",
    "order.cancel",
    orderId,
    message -> {
        message.getMessageProperties().setDelay(30 * 60 * 1000);  // 30 min
        return message;
    }
);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

version: "3.8"

services:
  # ============= RabbitMQ 节点 1 =============
  rabbitmq-1:
    image: rabbitmq:3.13-management
    container_name: rabbitmq-1
    hostname: rabbitmq-1
    ports:
      - "5672:5672"     # AMQP
      - "15672:15672"   # Management UI
      - "15692:15692"   # Prometheus Exporter
    environment:
      RABBITMQ_ERLANG_COOKIE: "secret-cookie-here"   # 节点间发现用
    volumes:
      - rabbitmq1-data:/var/lib/rabbitmq
    networks:
      - rabbitmq-net

  rabbitmq-2:
    image: rabbitmq:3.13-management
    container_name: rabbitmq-2
    hostname: rabbitmq-2
    ports:
      - "5673:5672"
      - "15673:15672"
      - "15693:15692"
    environment:
      RABBITMQ_ERLANG_COOKIE: "secret-cookie-here"
    volumes:
      - rabbitmq2-data:/var/lib/rabbitmq
    networks:
      - rabbitmq-net

  rabbitmq-3:
    image: rabbitmq:3.13-management
    container_name: rabbitmq-3
    hostname: rabbitmq-3
    ports:
      - "5674:5672"
      - "15674:15672"
      - "15694:15692"
    environment:
      RABBITMQ_ERLANG_COOKIE: "secret-cookie-here"
    volumes:
      - rabbitmq3-data:/var/lib/rabbitmq
    networks:
      - rabbitmq-net

  # ============= HAProxy 负载均衡 =============
  haproxy:
    image: haproxy:3.0
    container_name: rabbitmq-haproxy
    ports:
      - "5672:5672"      # 对外 AMQP 入口
    volumes:
      - ./haproxy.cfg:/usr/local/etc/haproxy/haproxy.cfg:ro
    depends_on:
      - rabbitmq-1
      - rabbitmq-2
      - rabbitmq-3
    networks:
      - rabbitmq-net

volumes:
  rabbitmq1-data:
  rabbitmq2-data:
  rabbitmq3-data:

networks:
  rabbitmq-net:
    driver: bridge

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

global
    log /dev/log local0
    maxconn 4096

defaults
    mode tcp
    log global
    timeout connect 5s
    timeout client 30s
    timeout server 30s

# AMQP 入口（4 层负载）
frontend rabbitmq-frontend
    bind *:5672
    default_backend rabbitmq-backend

backend rabbitmq-backend
    balance roundrobin
    option tcp-check
    tcp-check send-binary ff
    tcp-check expect binary 01
    server rabbitmq-1 rabbitmq-1:5672 check inter 5s
    server rabbitmq-2 rabbitmq-2:5672 check inter 5s
    server rabbitmq-3 rabbitmq-3:5672 check inter 5s

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

docker-compose up -d

# 1. 节点加入集群（在节点 2、3 上执行）
docker exec rabbitmq-2 rabbitmqctl stop_app
docker exec rabbitmq-2 rabbitmqctl join_cluster rabbit@rabbitmq-1
docker exec rabbitmq-2 rabbitmqctl start_app

# 2. 启用管理插件 + Quorum Queue 插件
docker exec rabbitmq-1 rabbitmq-plugins enable rabbitmq_management rabbitmq_prometheus

# 3. 验证集群状态
docker exec rabbitmq-1 rabbitmqctl cluster_status
# 应显示 3 节点

# 4. 声明 Quorum Queue
docker exec rabbitmq-1 rabbitmqadmin declare queue \
  name=order-created-quorum \
  durable=true \
  arguments='{"x-queue-type":"quorum","x-dead-letter-exchange":"order-dlx"}'

# 5. 访问 UI
# 浏览器打开 http://localhost:15672 (guest/guest)

1
2
3
4
5
6
7
8
9

# 演练 1：杀单节点
docker stop rabbitmq-2
# 观察：HAProxy 30 秒内剔除 rabbitmq-2，剩余 2 节点继续服务
# Quorum Queue 自动选新 Leader

# 演练 2：网络分区
docker network disconnect rabbitmq-net rabbitmq-3
# 观察：剩 2 节点满足 Raft 仲裁（半数 = 2），继续服务
# rabbitmq-3 重新接入后自动同步数据

1
2
3
4
5
6

Exchange: order-events (topic)
├─ routing key: order.created    → Queue: order-created (库存服务订阅)
├─ routing key: order.paid       → Queue: order-paid (财务服务订阅)
├─ routing key: order.shipped    → Queue: order-shipped (物流服务订阅)
├─ routing key: order.cancelled  → Queue: order-cancelled (客服订阅)
└─ routing key: order.#          → Queue: order-all (数仓订阅，全量数据)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

// 1. 定义统一接口
public interface MessageBus {
    void send(String topic, Object message);
    void sendWithTransaction(String topic, Object message, Runnable localTx);
}

// 2. RocketMQ 实现
@Service
@ConditionalOnProperty(name = "mq.type", havingValue = "rocketmq")
public class RocketMQBus implements MessageBus {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    // 实现接口
}

// 3. Kafka 实现
@Service
@ConditionalOnProperty(name = "mq.type", havingValue = "kafka")
public class KafkaBus implements MessageBus {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    // 实现接口
}

// 4. 业务代码注入接口
@Service
public class OrderService {
    @Autowired
    private MessageBus messageBus;  // 不关心是哪种 MQ
    public void createOrder(Order order) {
        orderRepository.save(order);
        messageBus.send("order-topic", order);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

spring:
  cloud:
    stream:
      bindings:
        order-out:
          destination: order-topic
          content-type: application/json
      rocketmq:
        binder:
          name-server: rocketmq-namesrv:9876