RabbitMQ 入门：架构解析、环境搭建与 Java 实战

工作流程与核心概念

RabbitMQ 的核心组件围绕 Broker（服务器）、Virtual Host（虚拟主机）、Exchange（交换机）、Queue（队列）、Connection（连接）和 Channel（信道）展开。

消息流转逻辑

1. Producer 建立 TCP Connection，并在其上创建 Channel。
2. 通过 Channel 将消息发送给 Exchange，指定 routing_key。
3. Exchange 根据类型（direct/topic/fanout/headers）和 Binding 规则，将消息路由到 0~N 个 Queue。
4. Queue 按 FIFO 缓冲消息；若开启持久化则落盘。
5. Consumer 建立 Connection/Channel，订阅或拉取 Queue 中的消息。
6. Consumer 处理消息后回复 ack，RabbitMQ 收到后删除消息；若未收到 ack 且断开，消息重新入队。

关键概念详解

• Virtual Host (vhost)：逻辑隔离单元，类似数据库中的库。不同 vhost 间的 Exchange 和 Queue 完全隔离。
• Exchange：消息入口，负责路由。不直接存储消息，而是根据规则转发。
• Queue：真正存储消息的内部对象。
• Connection & Channel：Connection 是 TCP 长连接，Channel 是其上的轻量级子连接。复用同一个 TCP 连接可显著减少开销，提高性能。
• Producer & Consumer：客户端角色，分别负责发布和接收消息。

Java 快速上手示例

以下示例基于 Java 客户端，展示从依赖引入到收发消息的完整流程。

1. 引入依赖

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

2. 生产者实现

建立连接与信道

我们需要配置 IP、端口、账号密码及虚拟主机。实际开发中建议从配置文件读取，切勿硬编码敏感信息。

ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("localhost"); // 替换为实际服务器 IP
connectionFactory.setPort(5672);
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
connectionFactory.setVirtualHost("/");

Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();

声明队列与发送消息

简单模式下，可使用默认交换机（空字符串）。注意队列参数的含义：

• durable: 是否持久化
• exclusive: 是否独占
• autoDelete: 无消费者时是否自动删除

// 声明队列
channel.queueDeclare("hello", true, false, true, null);

// 发送消息
String msg = "Hello RabbitMQ";
channel.basicPublish("", "hello", null, msg.getBytes());
System.out.println("[x] Sent '" + msg + "'");

释放资源

务必关闭信道和连接，防止资源泄漏。

channel.close();
connection.close();

完整代码参考：

package org.example.rabbitmq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ProducerDemo {
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");

        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare("hello", true, false, true, null);
        String msg = "Hello RabbitMQ";
        channel.basicPublish("", "hello", null, msg.getBytes());

        channel.close();
        connection.close();
    }
}

3. 消费者实现

消费者需要监听队列，接收到消息后执行回调。

package org.example.rabbitmq;

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ConsumerDemo {
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");

        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare("hello", true, false, true, null);

        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("[x] Received: '" + message + "'");
                // 业务逻辑处理...
            }
        };

        // autoAck 设为 true 表示自动确认，生产环境建议手动 ack
        channel.basicConsume("hello", true, consumer);
    }
}

运行消费者后，控制台应能打印出生产者发送的消息。至此，一个基础的点对点消息传递链路已打通。在实际生产中，还需关注异常处理、事务机制以及集群部署策略。