# 拉取帶管理界面的鏡像
docker pull rabbitmq:3.12-management
# 運(yùn)行容器（設(shè)置用戶名和密碼）
docker run -d \
  --name rabbitmq \
  -p 5672:5672 \
  -p 15672:15672 \
  -e RABBITMQ_DEFAULT_USER=admin \
  -e RABBITMQ_DEFAULT_PASS=password \
  rabbitmq:3.12-management

# 下載RabbitMQ
wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.8.35/rabbitmq-server-generic-unix-3.8.35.tar.xz
tar xvf rabbitmq-server-generic-unix-3.8.35.tar.xz
cd rabbitmq_server-3.8.35
# 啟動RabbitMQ（前臺啟動，方便查看日志）
./sbin/rabbitmq-server

Configuring logger redirection
##  ##      RabbitMQ 3.8.35
##  ## ##########  Copyright (c) 2007-2022 VMware, Inc. or its affiliates.
...（其他信息）
Starting broker... completed with 0 plugins.

# 檢查服務(wù)狀態(tài)
./sbin/rabbitmqctl status
# 預(yù)期輸出：
Status of node rabbit@localhost ...
Runtime OS PID: 1124
OS: Linux
Uptime (seconds): 30
Is under maintenance?: false

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.16.0</version>
</dependency>

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class RabbitMQProducer {
    private final static String QUEUE_NAME = "hello";
    public static void main(String[] args) throws Exception {
        // 1. 創(chuàng)建連接工廠
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("admin");
        factory.setPassword("password");
        // 2. 建立連接和通道
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            // 3. 聲明隊(duì)列（如果不存在則創(chuàng)建）
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            // 4. 發(fā)送消息
            String message = "Hello RabbitMQ!";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
        }
    }
}

import com.rabbitmq.client.*;
public class RabbitMQConsumer {
    private final static String QUEUE_NAME = "hello";
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("admin");
        factory.setPassword("password");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 聲明隊(duì)列（確保隊(duì)列存在）
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        // 創(chuàng)建回調(diào)處理器
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + message + "'");
            // 手動確認(rèn)消息（重要?。?
            try {
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            } catch (Exception e) {
                e.printStackTrace();
            }
        };
        // 開始消費(fèi)消息
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
    }
}

Producer → Exchange → Bindings → Queue → Consumer

// 精確匹配routing key
channel.exchangeDeclare("direct-exchange", BuiltinExchangeType.DIRECT);
channel.queueBind("queue1", "direct-exchange", "error");
channel.queueBind("queue2", "direct-exchange", "info");
// 消息會路由到queue1
channel.basicPublish("direct-exchange", "error", null, message.getBytes());

// 廣播到所有綁定的隊(duì)列
channel.exchangeDeclare("fanout-exchange", BuiltinExchangeType.FANOUT);
channel.queueBind("queue1", "fanout-exchange", "");
channel.queueBind("queue2", "fanout-exchange", "");
// 消息會路由到queue1和queue2
channel.basicPublish("fanout-exchange", "", null, message.getBytes());

// 基于模式匹配
channel.exchangeDeclare("topic-exchange", BuiltinExchangeType.TOPIC);
channel.queueBind("queue1", "topic-exchange", "*.orange.*");
channel.queueBind("queue2", "topic-exchange", "*.*.rabbit");
// 消息"quick.orange.rabbit"會路由到兩個隊(duì)列
channel.basicPublish("topic-exchange", "quick.orange.rabbit", null, message.getBytes());

// 啟用確認(rèn)模式
channel.confirmSelect();
// 異步確認(rèn)回調(diào)
channel.addConfirmListener(new ConfirmListener() {
    @Override
    public void handleAck(long deliveryTag, boolean multiple) {
        System.out.println("消息已確認(rèn): " + deliveryTag);
    }
    @Override
    public void handleNack(long deliveryTag, boolean multiple) {
        System.out.println("消息未確認(rèn)，需重發(fā): " + deliveryTag);
    }
});

// 手動確認(rèn)（推薦）
channel.basicConsume(queue, false, new DefaultConsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag,
                               Envelope envelope,
                               AMQP.BasicProperties properties,
                               byte[] body) throws IOException {
        // 處理消息
        processMessage(body);
        // 確認(rèn)消息（單個確認(rèn)）
        channel.basicAck(envelope.getDeliveryTag(), false);
    }
});

// 隊(duì)列持久化
boolean durable = true;
channel.queueDeclare("my-queue", durable, false, false, null);
// 消息持久化
AMQP.BasicProperties properties = MessageProperties.PERSISTENT_TEXT_PLAIN;
channel.basicPublish("", "my-queue", properties, message.getBytes());
// 交換機(jī)持久化
channel.exchangeDeclare("my-exchange", "direct", true);

// 定義死信交換機(jī)
channel.exchangeDeclare("dlx-exchange", "direct");
// 定義死信隊(duì)列
channel.queueDeclare("dlx-queue", true, false, false, null);
channel.queueBind("dlx-queue", "dlx-exchange", "dlx-routing-key");
// 創(chuàng)建普通隊(duì)列時指定死信配置
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", "dlx-exchange");
args.put("x-dead-letter-routing-key", "dlx-routing-key");
args.put("x-message-ttl", 60000); // 消息TTL 60秒
channel.queueDeclare("normal-queue", true, false, false, args);

// 創(chuàng)建優(yōu)先級隊(duì)列
Map<String, Object> args = new HashMap<>();
args.put("x-max-priority", 10); // 最高優(yōu)先級為10
channel.queueDeclare("priority-queue", true, false, false, args);
// 發(fā)送高優(yōu)先級消息
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
    .priority(5) // 優(yōu)先級5
    .build();
channel.basicPublish("", "priority-queue", properties, "High Priority Message".getBytes());

// 消費(fèi)者端設(shè)置預(yù)取數(shù)量為1（公平分發(fā)）
int prefetchCount = 1;
channel.basicQos(prefetchCount);

# 調(diào)整文件句柄限制
ulimit -n 65536
# 優(yōu)化Erlang VM參數(shù)
export RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="+P 1048576 +t 5000000"

# 加入集群
rabbitmqctl stop_app
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app
# 設(shè)置鏡像隊(duì)列策略（高可用）
rabbitmqctl set_policy ha-all ".*" '{"ha-mode":"all"}'

生產(chǎn)者 → [AMQP協(xié)議] → RabbitMQ Broker → [Exchange] → [Binding] → [Queue] → [Consumer]

// 1. 交換機(jī)持久化（必須?。?
channel.exchangeDeclare("my-exchange", "direct", true); // 第三個參數(shù)true
// 2. 隊(duì)列持久化（必須?。?
channel.queueDeclare("my-queue", true, false, false, null); // 第二個參數(shù)true
// 3. 消息持久化（必須?。?
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .deliveryMode(2) // 2=持久化，1=非持久化
    .build();
channel.basicPublish("my-exchange", "key", props, "message".getBytes());

// 開啟確認(rèn)模式（關(guān)鍵！）
channel.confirmSelect();
// 異步監(jiān)聽確認(rèn)結(jié)果
channel.addConfirmListener(new ConfirmListener() {
    @Override
    public void handleAck(long deliveryTag, boolean multiple) {
        System.out.println("? 消息已持久化: " + deliveryTag);
    }
    @Override
    public void handleNack(long deliveryTag, boolean multiple) {
        System.out.println("? 消息未確認(rèn)，需重發(fā): " + deliveryTag);
        // 重發(fā)邏輯（如：放入重試隊(duì)列）
    }
});

// 關(guān)鍵：設(shè)置autoAck=false（必須！）
channel.basicConsume("my-queue", false, (consumerTag, delivery) -> {
    try {
        // 處理消息（耗時操作）
        processMessage(delivery.getBody());
        // ? 手動確認(rèn)（必須！）
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    } catch (Exception e) {
        // 失敗時拒絕消息（可選：重新入隊(duì)）
        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
    }
}, consumerTag -> {});

# 設(shè)置鏡像策略（所有隊(duì)列都鏡像）
rabbitmqctl set_policy ha-all ".*" '{"ha-mode":"all"}'
# 檢查鏡像狀態(tài)
rabbitmqctl list_queues name mirrors

// 1. 持久化配置
channel.exchangeDeclare("ex", "direct", true);
channel.queueDeclare("queue", true, false, false, null);
channel.queueBind("queue", "ex", "key");
// 2. 生產(chǎn)者：持久化消息 + 確認(rèn)
channel.confirmSelect();
channel.basicPublish("ex", "key", 
    new AMQP.BasicProperties.Builder().deliveryMode(2).build(),
    "message".getBytes());
channel.waitForConfirmsOrDie(); // 等待確認(rèn)（同步阻塞）
// 3. 消費(fèi)者：手動ack
channel.basicConsume("queue", false, (tag, msg) -> {
    try {
        // 處理消息
        process(msg.getBody());
        channel.basicAck(msg.getEnvelope().getDeliveryTag(), false); // 必須手動ack
    } catch (Exception e) {
        channel.basicNack(msg.getEnvelope().getDeliveryTag(), false, true); // 重試
    }
}, tag -> {});

生產(chǎn)者 → RabbitMQ內(nèi)存隊(duì)列 → [消息寫入磁盤緩沖區(qū)] → [異步刷盤] → 磁盤文件

// 錯誤示例：未使用Confirm機(jī)制
channel.basicPublish(..., properties, "message".getBytes());
// 以為消息已持久化，其實(shí)可能還在內(nèi)存緩存中

// 1. 開啟Publisher Confirm（必須?。?
channel.confirmSelect();
// 2. 等待fsync完成（關(guān)鍵?。?
boolean confirmed = channel.waitForConfirmsOrDie(5000); // 等待5秒
// 3. 如果需要極致可靠（性能犧牲大），配置RabbitMQ
// 在rabbitmq.conf中添加：
# 每條消息后立即fsync（極端可靠，性能差）
disk_free_limit.absolute = 1GB
vm_memory_high_watermark.relative = 0.8
# 但不推薦，除非是金融系統(tǒng)

// 1. 持久化配置（三件套）
channel.exchangeDeclare("order-exchange", "direct", true);
channel.queueDeclare("order-queue", true, false, false, null);
channel.queueBind("order-queue", "order-exchange", "order");
// 2. 關(guān)鍵：使用Confirm機(jī)制 + 等待fsync
channel.confirmSelect();
channel.basicPublish("order-exchange", "order", 
    new AMQP.BasicProperties.Builder().deliveryMode(2).build(),
    "order-123".getBytes());
channel.waitForConfirmsOrDie(); // 等待消息真正落盤

import com.rabbitmq.client.*;
public class ReliableProducer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setAutomaticRecoveryEnabled(true); // 服務(wù)器斷連自動重連
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            // 1. 交換機(jī)持久化（必須）
            channel.exchangeDeclare("payment-ex", "direct", true);
            // 2. 隊(duì)列持久化（必須）
            channel.queueDeclare("payment-queue", true, false, false, null);
            // 3. 隊(duì)列綁定（必須）
            channel.queueBind("payment-queue", "payment-ex", "order");
            // 4. 開啟 Publisher Confirms（必須?。?
            channel.confirmSelect();
            // 5. 發(fā)送持久化消息（deliveryMode=2）
            AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                    .deliveryMode(2) // 2 = 持久化
                    .contentType("application/json")
                    .build();
            // 6. 關(guān)鍵：重試邏輯 + 等待 fsync 完成
            String message = "{\"orderId\":\"1001\",\"amount\":100.00}";
            channel.basicPublish("payment-ex", "order", props, message.getBytes());
            // 等待 fsync 完成（RabbitMQ 保證此時消息已寫入物理磁盤）
            if (!channel.waitForConfirms(5000)) { // 5秒超時
                System.err.println("消息發(fā)送超時，觸發(fā)重試！");
                // 重試邏輯（實(shí)際項(xiàng)目建議用指數(shù)退避）
                retryPublish(channel, "payment-ex", "order", props, message);
            }
        }
    }
    private static void retryPublish(Channel channel, String exchange, String routingKey,
                                    AMQP.BasicProperties props, String message) throws Exception {
        // 實(shí)際項(xiàng)目建議：指數(shù)退避重試（避免風(fēng)暴）
        for (int i = 0; i < 3; i++) {
            try {
                channel.basicPublish(exchange, routingKey, props, message.getBytes());
                if (channel.waitForConfirms(5000)) {
                    return;
                }
            } catch (Exception e) {
                Thread.sleep(100 * (i + 1)); // 100ms, 200ms, 400ms
            }
        }
        throw new RuntimeException("重試3次仍失敗，消息丟失！");
    }
}

public class ReliableConsumer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            // 1. 聲明持久化交換機(jī)/隊(duì)列（必須與發(fā)送方一致）
            channel.exchangeDeclare("payment-ex", "direct", true);
            channel.queueDeclare("payment-queue", true, false, false, null);
            channel.queueBind("payment-queue", "payment-ex", "order");
            // 2. 關(guān)閉自動 ACK（必須?。?
            channel.basicConsume("payment-queue", false, 
                (consumerTag, delivery) -> {
                    try {
                        String message = new String(delivery.getBody(), "UTF-8");
                        System.out.println("收到消息: " + message);
                        // 3. 業(yè)務(wù)處理（模擬耗時操作）
                        processPayment(message);
                        // 4. 成功處理后手動 ACK（必須?。?
                        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                    } catch (Exception e) {
                        System.err.println("處理失敗，觸發(fā)重試: " + e.getMessage());
                        // 5. 重試邏輯：拒絕消息（RabbitMQ 會重新投遞）
                        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
                    }
                },
                consumerTag -> {});
            System.out.println("消費(fèi)者已啟動，等待消息...");
            Thread.sleep(Long.MAX_VALUE); // 保持運(yùn)行
        }
    }
    private static void processPayment(String message) throws Exception {
        // 模擬業(yè)務(wù)處理（如支付接口調(diào)用）
        if (Math.random() > 0.9) { // 10% 模擬失敗
            throw new RuntimeException("支付失敗");
        }
        System.out.println("支付成功處理完成");
    }
}

# 1. 磁盤空間（避免因磁盤滿導(dǎo)致消息丟失）
disk_free_limit.absolute = 1GB
vm_memory_high_watermark.relative = 0.8
# 2. 持久化優(yōu)化（默認(rèn)已啟用，確保異步刷盤安全）
# 無需額外配置，但需確認(rèn)
# file_handle_cache_size = 1024
# disk_free_limit.absolute = 1GB
# 3. 高可用集群（關(guān)鍵！避免單點(diǎn)故障）
# 以下配置在集群節(jié)點(diǎn)上統(tǒng)一設(shè)置
cluster_formation.peer_discovery_implementation = rabbit_peer_discovery_classic_config
cluster_formation.classic_config.nodes.1 = rabbit@node1
cluster_formation.classic_config.nodes.2 = rabbit@node2
cluster_formation.classic_config.nodes.3 = rabbit@node3
# 4. 鏡像隊(duì)列（確保隊(duì)列在多個節(jié)點(diǎn)有副本）
# 以下配置在管理界面或命令行設(shè)置
# rabbitmqctl set_policy ha-all "^" '{"ha-mode":"all"}'

# RabbitMQ 服務(wù)器配置 (rabbitmq.conf)
disk_free_limit.absolute = 1GB
vm_memory_high_watermark.relative = 0.8
# 鏡像隊(duì)列配置（命令行）
rabbitmqctl set_policy ha-all "^" '{"ha-mode":"all"}'
# 生產(chǎn)者代碼（Java）
channel.confirmSelect();
channel.basicPublish(...);
channel.waitForConfirmsOrDie(); # 5秒超時
# 消費(fèi)者代碼（Java）
channel.basicConsume(..., false); # 關(guān)閉自動 ACK
// 處理成功：channel.basicAck(...)
// 處理失?。篶hannel.basicNack(..., true) # 重投