RabbitMQ从入门到原理再到实战应用

# 拉取带管理界面的镜像
docker pull rabbitmq:3.12-management
# 运行容器（设置用户名和密码）
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.

# 检查服务状态
./sbin/rabbitmqctl status
# 预期输出：
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. 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("admin");
        factory.setPassword("password");
        // 2. 建立连接和通道
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            // 3. 声明队列（如果不存在则创建）
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            // 4. 发送消息
            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();
        // 声明队列（确保队列存在）
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        // 创建回调处理器
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + message + "'");
            // 手动确认消息（重要！）
            try {
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            } catch (Exception e) {
                e.printStackTrace();
            }
        };
        // 开始消费消息
        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());

// 广播到所有绑定的队列
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"会路由到两个队列
channel.basicPublish("topic-exchange", "quick.orange.rabbit", null, message.getBytes());

// 启用确认模式
channel.confirmSelect();
// 异步确认回调
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("消息未确认，需重发: " + deliveryTag);
    }
});

// 手动确认（推荐）
channel.basicConsume(queue, false, new DefaultConsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag,
                               Envelope envelope,
                               AMQP.BasicProperties properties,
                               byte[] body) throws IOException {
        // 处理消息
        processMessage(body);
        // 确认消息（单个确认）
        channel.basicAck(envelope.getDeliveryTag(), false);
    }
});

// 队列持久化
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());
// 交换机持久化
channel.exchangeDeclare("my-exchange", "direct", true);

// 定义死信交换机
channel.exchangeDeclare("dlx-exchange", "direct");
// 定义死信队列
channel.queueDeclare("dlx-queue", true, false, false, null);
channel.queueBind("dlx-queue", "dlx-exchange", "dlx-routing-key");
// 创建普通队列时指定死信配置
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);

// 创建优先级队列
Map<String, Object> args = new HashMap<>();
args.put("x-max-priority", 10); // 最高优先级为10
channel.queueDeclare("priority-queue", true, false, false, args);
// 发送高优先级消息
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
    .priority(5) // 优先级5
    .build();
channel.basicPublish("", "priority-queue", properties, "High Priority Message".getBytes());

// 消费者端设置预取数量为1（公平分发）
int prefetchCount = 1;
channel.basicQos(prefetchCount);

# 调整文件句柄限制
ulimit -n 65536
# 优化Erlang VM参数
export RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="+P 1048576 +t 5000000"

# 加入集群
rabbitmqctl stop_app
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app
# 设置镜像队列策略（高可用）
rabbitmqctl set_policy ha-all ".*" '{"ha-mode":"all"}'

生产者 → [AMQP协议] → RabbitMQ Broker → [Exchange] → [Binding] → [Queue] → [Consumer]

// 1. 交换机持久化（必须！）
channel.exchangeDeclare("my-exchange", "direct", true); // 第三个参数true
// 2. 队列持久化（必须！）
channel.queueDeclare("my-queue", true, false, false, null); // 第二个参数true
// 3. 消息持久化（必须！）
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .deliveryMode(2) // 2=持久化，1=非持久化
    .build();
channel.basicPublish("my-exchange", "key", props, "message".getBytes());

// 开启确认模式（关键！）
channel.confirmSelect();
// 异步监听确认结果
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("❌ 消息未确认，需重发: " + deliveryTag);
        // 重发逻辑（如：放入重试队列）
    }
});

// 关键：设置autoAck=false（必须！）
channel.basicConsume("my-queue", false, (consumerTag, delivery) -> {
    try {
        // 处理消息（耗时操作）
        processMessage(delivery.getBody());
        // ✅ 手动确认（必须！）
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    } catch (Exception e) {
        // 失败时拒绝消息（可选：重新入队）
        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
    }
}, consumerTag -> {});

# 设置镜像策略（所有队列都镜像）
rabbitmqctl set_policy ha-all ".*" '{"ha-mode":"all"}'
# 检查镜像状态
rabbitmqctl list_queues name mirrors

// 生产者：未持久化 + 自动确认
channel.basicPublish("", "queue", null, "message".getBytes());
// 消费者：autoAck=true（致命错误！）
channel.basicConsume("queue", true, ...);

// 1. 持久化配置
channel.exchangeDeclare("ex", "direct", true);
channel.queueDeclare("queue", true, false, false, null);
channel.queueBind("queue", "ex", "key");
// 2. 生产者：持久化消息 + 确认
channel.confirmSelect();
channel.basicPublish("ex", "key", 
    new AMQP.BasicProperties.Builder().deliveryMode(2).build(),
    "message".getBytes());
channel.waitForConfirmsOrDie(); // 等待确认（同步阻塞）
// 3. 消费者：手动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 -> {});

生产者 → RabbitMQ内存队列 → [消息写入磁盘缓冲区] → [异步刷盘] → 磁盘文件

// 错误示例：未使用Confirm机制
channel.basicPublish(..., properties, "message".getBytes());
// 以为消息已持久化，其实可能还在内存缓存中

// 1. 开启Publisher Confirm（必须！）
channel.confirmSelect();
// 2. 等待fsync完成（关键！）
boolean confirmed = channel.waitForConfirmsOrDie(5000); // 等待5秒
// 3. 如果需要极致可靠（性能牺牲大），配置RabbitMQ
// 在rabbitmq.conf中添加：
# 每条消息后立即fsync（极端可靠，性能差）
disk_free_limit.absolute = 1GB
vm_memory_high_watermark.relative = 0.8
# 但不推荐，除非是金融系统

// 1. 持久化配置（三件套）
channel.exchangeDeclare("order-exchange", "direct", true);
channel.queueDeclare("order-queue", true, false, false, null);
channel.queueBind("order-queue", "order-exchange", "order");
// 2. 关键：使用Confirm机制 + 等待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); // 服务器断连自动重连
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            // 1. 交换机持久化（必须）
            channel.exchangeDeclare("payment-ex", "direct", true);
            // 2. 队列持久化（必须）
            channel.queueDeclare("payment-queue", true, false, false, null);
            // 3. 队列绑定（必须）
            channel.queueBind("payment-queue", "payment-ex", "order");
            // 4. 开启 Publisher Confirms（必须！）
            channel.confirmSelect();
            // 5. 发送持久化消息（deliveryMode=2）
            AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                    .deliveryMode(2) // 2 = 持久化
                    .contentType("application/json")
                    .build();
            // 6. 关键：重试逻辑 + 等待 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("消息发送超时，触发重试！");
                // 重试逻辑（实际项目建议用指数退避）
                retryPublish(channel, "payment-ex", "order", props, message);
            }
        }
    }
    private static void retryPublish(Channel channel, String exchange, String routingKey,
                                    AMQP.BasicProperties props, String message) throws Exception {
        // 实际项目建议：指数退避重试（避免风暴）
        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. 声明持久化交换机/队列（必须与发送方一致）
            channel.exchangeDeclare("payment-ex", "direct", true);
            channel.queueDeclare("payment-queue", true, false, false, null);
            channel.queueBind("payment-queue", "payment-ex", "order");
            // 2. 关闭自动 ACK（必须！）
            channel.basicConsume("payment-queue", false, 
                (consumerTag, delivery) -> {
                    try {
                        String message = new String(delivery.getBody(), "UTF-8");
                        System.out.println("收到消息: " + message);
                        // 3. 业务处理（模拟耗时操作）
                        processPayment(message);
                        // 4. 成功处理后手动 ACK（必须！）
                        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                    } catch (Exception e) {
                        System.err.println("处理失败，触发重试: " + e.getMessage());
                        // 5. 重试逻辑：拒绝消息（RabbitMQ 会重新投递）
                        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
                    }
                },
                consumerTag -> {});
            System.out.println("消费者已启动，等待消息...");
            Thread.sleep(Long.MAX_VALUE); // 保持运行
        }
    }
    private static void processPayment(String message) throws Exception {
        // 模拟业务处理（如支付接口调用）
        if (Math.random() > 0.9) { // 10% 模拟失败
            throw new RuntimeException("支付失败");
        }
        System.out.println("支付成功处理完成");
    }
}

# 1. 磁盘空间（避免因磁盘满导致消息丢失）
disk_free_limit.absolute = 1GB
vm_memory_high_watermark.relative = 0.8
# 2. 持久化优化（默认已启用，确保异步刷盘安全）
# 无需额外配置，但需确认
# file_handle_cache_size = 1024
# disk_free_limit.absolute = 1GB
# 3. 高可用集群（关键！避免单点故障）
# 以下配置在集群节点上统一设置
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. 镜像队列（确保队列在多个节点有副本）
# 以下配置在管理界面或命令行设置
# rabbitmqctl set_policy ha-all "^" '{"ha-mode":"all"}'

# RabbitMQ 服务器配置 (rabbitmq.conf)
disk_free_limit.absolute = 1GB
vm_memory_high_watermark.relative = 0.8
# 镜像队列配置（命令行）
rabbitmqctl set_policy ha-all "^" '{"ha-mode":"all"}'
# 生产者代码（Java）
channel.confirmSelect();
channel.basicPublish(...);
channel.waitForConfirmsOrDie(); # 5秒超时
# 消费者代码（Java）
channel.basicConsume(..., false); # 关闭自动 ACK
// 处理成功：channel.basicAck(...)
// 处理失败：channel.basicNack(..., true) # 重投