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) # 重投