1
// 声明 Exchange
2
channel.exchangeDeclare("order-direct", "direct", true);   // 持久化
3
channel.exchangeDeclare("log-fanout", "fanout", true);
4
channel.exchangeDeclare("event-topic", "topic", true);
5
channel.exchangeDeclare="route-headers", "headers", true);

1
// 声明 Direct Exchange 和绑定
2
channel.exchangeDeclare("order-exchange", "direct", true);
3

4
// 声明队列
5
channel.queueDeclare("new-orders", true, false, false, null);
6
channel.queueDeclare("paid-orders", true, false, false, null);
7
channel.queueDeclare("cancel-orders", true, false, false, null);
8

9
// 绑定：routing key 精确匹配
10
channel.queueBind("new-orders", "order-exchange", "order.new");
11
channel.queueBind("paid-orders", "order-exchange", "order.paid");
12
channel.queueBind("cancel-orders", "order-exchange", "order.cancel");
13

14
// 发送消息
15
channel.basicPublish("order-exchange", "order.new", null, orderJson.getBytes());

1
// 声明 Fanout Exchange
2
channel.exchangeDeclare("log-fanout", "fanout", true);
3

4
// 多个队列绑定到同一个 Fanout Exchange
5
channel.queueDeclare("log-file", true, false, false, null);
6
channel.queueDeclare("log-elasticsearch", true, false, false, null);
7
channel.queueDeclare("log-monitor", true, false, false, null);
8

9
// 绑定：无需 routing key
10
channel.queueBind("log-file", "log-fanout", "");
11
channel.queueBind("log-elasticsearch", "log-fanout", "");
12
channel.queueBind("log-monitor", "log-fanout", "");
13

14
// 发送消息——所有队列都会收到
15
channel.basicPublish("log-fanout", "", null, logData.getBytes());

1
// 声明 Topic Exchange
2
channel.exchangeDeclare("event-topic", "topic", true);
3

4
// 绑定：不同消费者订阅不同模式
5
channel.queueBind("all-orders", "event-topic", "order.#");        // 所有订单事件
6
channel.queueBind("new-orders", "event-topic", "order.new");      // 仅新订单
7
channel.queueBind("vip-orders", "event-topic", "order.*.vip");    // VIP 订单
8
channel.queueBind("all-events", "event-topic", "#");              // 所有事件
9

10
// 发送消息
11
channel.basicPublish("event-topic", "order.new.vip", null, eventJson.getBytes());
12
// 匹配：all-orders（order.#）、vip-orders（order.*.vip）、all-events（#）

1
// 声明 Headers Exchange
2
channel.exchangeDeclare("route-headers", "headers", true);
3

4
// 绑定：匹配条件
5
Map<String, Object> bindingArgs1 = new HashMap<>();
6
bindingArgs1.put("x-match", "all");     // 所有头都匹配
7
bindingArgs1.put("format", "pdf");
8
bindingArgs1.put("type", "report");
9

10
Map<String, Object> bindingArgs2 = new HashMap<>();
11
bindingArgs2.put("x-match", "any");     // 任一头匹配
12
bindingArgs2.put("format", "pdf");
13
bindingArgs2.put("format", "csv");
14

15
channel.queueBind("pdf-reports", "route-headers", "", bindingArgs1);
16
channel.queueBind("data-files", "route-headers", "", bindingArgs2);
17

18
// 发送消息
19
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
20
    .headers(Map.of("format", "pdf", "type", "report"))
21
    .build();
22
channel.basicPublish("route-headers", "", props, data.getBytes());

1
// 声明 Classic Queue
2
channel.queueDeclare("classic-queue", true, false, false, null);
3

4
// 声明 Quorum Queue（通过参数指定）
5
Map<String, Object> quorumArgs = new HashMap<>();
6
quorumArgs.put("x-queue-type", "quorum");
7
quorumArgs.put("x-delivery-limit", 3);  // 投递次数限制
8
channel.queueDeclare("quorum-queue", true, false, false, quorumArgs);
9

10
// 声明 Stream Queue
11
Map<String, Object> streamArgs = new HashMap<>();
12
streamArgs.put("x-queue-type", "stream");
13
streamArgs.put("x-max-length-bytes", 1073741824L);  // 1GB
14
streamArgs.put("x-stream-max-segment-size-bytes", 67108864L);  // 64MB
15
channel.queueDeclare("event-stream", true, false, false, streamArgs);

1
# 查看 Quorum Queue 状态
2
rabbitmq-ctl list_queues name type quorum_status leader members
3

4
# 输出示例：
5
# quorum-queue  quorum  running  rabbit@node1  [rabbit@node1, rabbit@node2, rabbit@node3]

1
// Stream 消费：支持偏移量消费
2
channel.basicQos(100);  // 预取
3

4
// 从指定偏移量开始消费
5
Map<String, Object> consumeArgs = new HashMap<>();
6
consumeArgs.put("x-stream-offset", "first");     // 从第一条开始
7
// consumeArgs.put("x-stream-offset", "last");    // 从最新开始
8
// consumeArgs.put("x-stream-offset", 12345L);    // 从指定偏移量开始
9
// consumeArgs.put("x-stream-offset", timestamp); // 从指定时间开始
10

11
channel.basicConsume("event-stream", false, consumeArgs, consumer);

1
// 方式一：事务（不推荐，性能差）
2
channel.txSelect();  // 开启事务
3
try {
4
    channel.basicPublish("order-exchange", "order.new", null, data.getBytes());
5
    channel.txCommit();  // 提交
6
} catch (Exception e) {
7
    channel.txRollback();  // 回滚
8
}
9

10
// 方式二：Publisher Confirm（推荐）
11
channel.confirmSelect();  // 开启确认模式
12

13
// 异步确认回调
14
channel.addConfirmListener(new ConfirmListener() {
15
    @Override
16
    public void handleAck(long deliveryTag, boolean multiple) {
17
        // 消息确认成功
18
        System.out.println("Message confirmed: " + deliveryTag);
19
    }
20

21
    @Override
22
    public void handleNack(long deliveryTag, boolean multiple) {
23
        // 消息确认失败，需要重发
24
        System.out.println("Message rejected: " + deliveryTag);
25
        resendMessage(deliveryTag);
26
    }
27
});
28

29
// 发送消息
30
channel.basicPublish("order-exchange", "order.new", null, data.getBytes());
31

32
// 批量确认
33
channel.waitForConfirmsOrDie(5000);  // 最多等待 5 秒

1
// 手动确认模式（推荐）
2
boolean autoAck = false;
3
channel.basicConsume("order-queue", autoAck, new DefaultConsumer(channel) {
4
    @Override
5
    public void handleDelivery(String consumerTag, Envelope envelope,
6
                               AMQP.BasicProperties properties, byte[] body) {
7
        try {
8
            processOrder(body);
9
            // 处理成功，确认消息
10
            channel.basicAck(envelope.getDeliveryTag(), false);
11
        } catch (Exception e) {
12
            // 处理失败，拒绝消息并重新入队
13
            channel.basicNack(envelope.getDeliveryTag(), false, true);
14
            // 或拒绝不重新入队（进入死信队列）
15
            // channel.basicReject(envelope.getDeliveryTag(), false);
16
        }
17
    }
18
});

1
// 声明带死信的队列
2
Map<String, Object> queueArgs = new HashMap<>();
3
queueArgs.put("x-dead-letter-exchange", "dlx-exchange");       // 死信 Exchange
4
queueArgs.put("x-dead-letter-routing-key", "dead.order");      // 死信 routing key
5
queueArgs.put("x-message-ttl", 60000);                         // 消息 TTL 60 秒
6
queueArgs.put("x-max-length", 10000);                          // 队列最大长度
7

8
channel.queueDeclare("order-queue", true, false, false, queueArgs);
9

10
// 声明死信 Exchange 和队列
11
channel.exchangeDeclare("dlx-exchange", "direct", true);
12
channel.queueDeclare("dead-letter-queue", true, false, false, null);
13
channel.queueBind("dead-letter-queue", "dlx-exchange", "dead.order");

1
// 死信处理器
2
channel.basicConsume("dead-letter-queue", false, new DefaultConsumer(channel) {
3
    @Override
4
    public void handleDelivery(String consumerTag, Envelope envelope,
5
                               AMQP.BasicProperties properties, byte[] body) {
6
        // 获取死信原因
7
        String reason = properties.getHeaders().get("x-death").toString();
8
        int retryCount = getRetryCount(properties);
9

10
        if (retryCount < 3) {
11
            // 重试：重新发送到业务 Exchange
12
            channel.basicPublish("order-exchange", "order.new",
13
                properties, body);
14
        } else {
15
            // 超过重试次数：记录到数据库并发送告警
16
            saveToDatabase(body, reason);
17
            sendAlert("Dead letter exceeded retry limit", body);
18
        }
19

20
        channel.basicAck(envelope.getDeliveryTag(), false);
21
    }
22
});

1
# 加入集群
2
rabbitmqctl stop_app
3
rabbitmqctl join_cluster rabbit@node1
4
rabbitmqctl start_app
5

6
# 查看集群状态
7
rabbitmqctl cluster_status
8

9
# 查看队列分布
10
rabbitmqctl list_queues name pid owner_pid

1
# 声明 Quorum Queue（3 副本）
2
# 通过策略设置
3
rabbitmqctl set_policy ha-quorum ".*" '{"queue-type":"quorum"}' --apply-to queues
4

5
# 查看 Quorum Queue 状态
6
rabbitmq-ctl list_queues name type quorum_status leader members
7

8
# 手动触发 Leader 切换
9
rabbitmq-ctl force_leader_of_quorum_queue quorum-queue

1
// 批量发送优化
2
channel.confirmSelect();
3
int batchSize = 100;
4

5
for (int i = 0; i < 10000; i++) {
6
    channel.basicPublish("exchange", "routing.key", null, message.getBytes());
7
    if ((i + 1) % batchSize == 0) {
8
        channel.waitForConfirmsOrDie(5000);  // 每 100 条确认一次
9
    }
10
}

1
// 预取优化
2
// prefetchCount = 1：严格轮询，公平但慢
3
// prefetchCount = 10-100：平衡吞吐和公平
4
// prefetchCount = 0：无限制，快但可能不均衡
5
channel.basicQos(50);  // 推荐 50-100

1
// 惰性队列：消息直接写磁盘，减少内存占用
2
Map<String, Object> lazyArgs = new HashMap<>();
3
lazyArgs.put("x-queue-mode", "lazy");
4
channel.queueDeclare("large-queue", true, false, false, lazyArgs);