RabbitMQ - 1 ( 7000 字 RabbitMQ 入门级教程 )

一：

在互联网行业，许多公司喜欢用动物命名产品或作为公司的 Logo 和吉祥物，比如腾讯的企鹅、京东的狗、美团的袋鼠、携程的海豚，而阿里更是凭借蚂蚁、飞猪、天猫、菜鸟、闲鱼、盒马等，打造了一座“动物园”。Rabbit（兔子）也是一家公司的名字，其产品 RabbitMQ 是一个消息队列（Message Queue，MQ）服务，实现了 AMQP（高级消息队列协议），并成为当前主流的消息中间件之一。

1.1 什么是 MQ

MQ（Message Queue，消息队列）本质上是一个队列，遵循 FIFO（先入先出）原则，只不过队列中存放的是消息（message）。消息可以非常简单，比如仅包含文本字符串或 JSON 格式，也可以较为复杂，例如内嵌对象。MQ 通常用于分布式系统之间的通信，而系统之间的调用一般有两种方式：

1.2 MQ 的作用

RabbitMQ 是一种 MQ 的实现，其主要功能是接收和转发消息。在不同的应用场景中，MQ 可以发挥多种作用。

1.3 为什么学习 RabbitMQ

目前业界有多种 MQ 产品可供选择，例如 RabbitMQ、RocketMQ、ActiveMQ、Kafka 和 ZeroMQ 等，甚至也有直接使用 Redis 作为消息队列的案例。这些消息队列各有侧重，没有绝对的优劣之分，只有适不适合。在实际选型时，需要根据自身需求和 MQ 产品的特点进行综合考量。

由于 RabbitMQ 综合能力较强，且我们的项目并非超高并发，对 RabbitMQ 社区的成熟度和管理界面的友好性需求较高，因此接下来主要学习 RabbitMQ 的使用。

二： RabbitMQ 核心概念

RabbitMQ是⼀个消息中间件也是⼀个⽣产者消费者模型，它负责接收、存储并转发消息.，界面上的导航栏分为 6 部分，每部分都有其特定的功能含义。在深入了解之前，我们先来看看 RabbitMQ 的工作流程，以便更好地理解各部分的作用。

2.1 Producer 和 Consumer

生产者（Producer）负责创建消息并将其发布到 RabbitMQ 中。在实际应用中，消息通常是具有一定业务逻辑结构的数据，例如 JSON 字符串。消息可以附带标签，RabbitMQ 会根据这些标签进行路由，将消息发送给感兴趣的消费者（Consumer）。

消费者通过连接 RabbitMQ 服务器来消费消息。在消费过程中，消息的标签会被丢弃，消费者只接收消息内容，并不会知道消息的生产者是谁，也无需关心消息的来源。对于 RabbitMQ 而言，一个 Broker 可以理解为一个 RabbitMQ 服务节点或实例，大多数情况下可视为一台 RabbitMQ 服务器。

2.2 Connection 和 Channel

通道的主要作用是将消息的读写操作复用到同一个 TCP 连接上，减少连接的建立和关闭开销，从而提高系统性能。

2.3 Virtual host

Virtual host（虚拟主机）是一个逻辑概念，用于为消息队列提供隔离机制。在 RabbitMQ 中，一个 Broker Server 上可以包含多个 Virtual Host，不同用户可以通过划分 vhost 来独立使用 RabbitMQ 服务，在各自的 vhost 中创建和管理 exchange、queue 等资源。它类似于 MySQL 中的 “database” 概念，是逻辑上的集合，而非类似 VMware 那样的物理虚拟机。

2.4 Queue

Queue 是 RabbitMQ 的内部对象，用于存储消息。多个消费者可以同时订阅同一个队列，共同消费其中的消息。

2.5 Exchange

Exchange（交换机）是消息到达 RabbitMQ Broker 的第一站，负责接收生产者发送的消息，并根据特定的规则将这些消息路由到一个或多个队列中。交换机起到消息路由的作用，通过其类型和规则来决定如何转发消息。类似于快递物流，根据收件地址分派快递到不同的站点，再由站点配送到收件人手中，交换机的作用就相当于完成这一步的分配工作。

2.6 RabbitMQ 工作流程

在理解了上述概念之后，我们可以通过回顾这张图，进一步梳理 RabbitMQ 的工作流程。

1. Producer（生产者）生成一条消息。
2. Producer 连接到 RabbitMQ Broker，建立一个连接（Connection）并开启一个信道（Channel）。
3. Producer 声明一个交换机（Exchange），用于路由消息。
4. Producer 声明一个队列（Queue），用于存储消息。
5. Producer 将消息发送到 RabbitMQ Broker。
6. RabbitMQ Broker 接收消息并将其存入相应的队列（Queue）。如果找不到对应的队列，则根据生产者的配置选择丢弃消息或退回给生产者。

2.7 AMQP

AMQP（Advanced Message Queuing Protocol，高级消息队列协议）定义了一套完整的消息交换功能，包括交换器（Exchange）和队列（Queue）等组件。这些组件协同工作，使生产者能够将消息发送到交换器，随后由队列接收并存储，等待消费者取用。此外，AMQP 还定义了一个网络协议，用于支持客户端应用与消息代理之间的交互通信。

RabbitMQ 是 AMQP 协议的实现，使用 Erlang 语言开发。换句话说，RabbitMQ 遵循 AMQP 的规范，同时还支持其他协议，如 STOMP 和 MQTT。由于 RabbitMQ 的模型结构与 AMQP 的模型结构一致，它可以充分发挥 AMQP 的强大功能。

三：web 界面操作

RabbitMQ 管理界面中的 Connections、Channels、Exchange 和 Queues 对应的概念与上述流程图中的内容一致。其中，Overview 表示系统的概览视图，Admin 则用于用户和权限管理。在操作 RabbitMQ 前需要先创建一个 Virtual Host。接下来，我们将详细介绍具体的操作步骤。

3.1 用户相关操作

3.1.1 添加用户

1. 点击 Admin -> Add user

2. 设置账号密码及权限

3. 观察用户是否添加成功

3.1.2 删除用户

1. 点击要删除的用户, 查看用户详情

2. 在用户详情页面, 进行更新或删除操作

   • 设置对虚拟机的操作权限

   • 更新 / 删除用户
     

3.1.3 退出当前用户

3.2 虚拟主机相关操作

3.2.1 创建虚拟主机

在 Admin 标签页下，点击右侧的 Virtual Hosts -> Add a new virtual host，然后设置虚拟主机的名称。

四： RabbitMQ 快速入门

1. 引入依赖

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

2. 编写生产者代码

public class RabbitProducer {
    public static void main(String[] args) throws Exception {
        // 1. 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();

        // 2. 设置连接参数
        factory.setHost("110.41.51.65"); // 设置主机地址（默认值为localhost）
        factory.setPort(15673);         // 设置端口号（默认值为5672）
        factory.setVirtualHost("bite"); // 设置虚拟主机名称（默认值为 "/"）
        factory.setUsername("study");  // 设置用户名（默认值为guest）
        factory.setPassword("study");  // 设置密码（默认值为guest）

        // 3. 创建连接
        Connection connection = factory.newConnection();

        // 4. 创建通道
        Channel channel = connection.createChannel();

        // 5. 声明队列
        /*
         * 1. queue: 队列名称
         * 2. durable: 是否持久化（true：队列会持久化到磁盘，MQ重启后队列仍在）
         * 3. exclusive: 是否独占（true：队列只允许当前连接使用，连接关闭后队列自动删除）
         * 4. autoDelete: 是否自动删除（true：当队列中没有消息且没有消费者时自动删除）
         * 5. arguments: 队列的其他参数（如 TTL、消息长度限制等）
         *
         * 如果指定的队列不存在，则会自动创建该队列；如果队列已存在，则使用已有队列。
         */
        channel.queueDeclare("hello", true, false, false, null);

        // 6. 发送消息到队列
        /*
         * 1. exchange: 交换机名称（简单模式下使用默认的""交换机）
         * 2. routingKey: 路由键（此处设置为队列名称，决定消息发送到哪个队列）
         * 3. props: 配置信息（如消息的优先级、过期时间等，可设为null）
         * 4. body: 消息内容，字节数组形式
         *
         * 使用默认的""交换机时，routingKey 必须与队列名称一致，才能正确路由消息到指定队列。
         */
        String msg = "Hello World"; // 消息内容
        channel.basicPublish("", "hello", null, msg.getBytes()); // 发送消息

        System.out.println(msg + " 消息发送成功");

        // 7. 释放资源
        channel.close();    // 关闭通道
        connection.close(); // 关闭连接
    }
}

3. 编写消费者代码

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

public class RabbitmqConsumer {
    public static void main(String[] args) throws Exception {
        // 1. 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();

        // 2. 设置连接参数
        factory.setHost("110.41.51.65"); // 设置主机地址（默认值为 localhost）
        factory.setPort(15673);         // 设置端口号（默认值为 5672）
        factory.setVirtualHost("bite"); // 设置虚拟主机名称（默认值为 "/"）
        factory.setUsername("study");  // 设置用户名（默认值为 guest）
        factory.setPassword("study");  // 设置密码（默认值为 guest）

        // 3. 创建连接
        Connection connection = factory.newConnection();

        // 4. 创建通道
        Channel channel = connection.createChannel();

        // 5. 声明队列
        /*
         * 1. queue: 队列名称
         * 2. durable: 是否持久化（true：队列会持久化到磁盘，MQ 重启后队列仍在）
         * 3. exclusive: 是否独占（true：队列只允许当前连接使用，连接关闭后队列自动删除）
         * 4. autoDelete: 是否自动删除（true：当队列中没有消息且没有消费者时自动删除）
         * 5. arguments: 队列的其他参数（如 TTL、消息长度限制等）
         *
         * 如果指定的队列不存在，则会自动创建该队列；如果队列已存在，则使用已有队列。
         */
        channel.queueDeclare("hello", true, false, false, null);

        // 6. 接收并消费消息
        /*
         * 1. queue: 队列名称
         * 2. autoAck: 是否自动确认（true：消费者收到消息后自动告诉 MQ 消息已消费）
         * 3. callback: 回调对象，处理接收到的消息
         */
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /*
             * 回调方法，当收到消息后会自动执行该方法
             * 参数说明：
             * 1. consumerTag: 消费者标签（标识消费者）
             * 2. envelope: 包含一些信息，如交换机、路由键
             * 3. properties: 消息的配置信息
             * 4. body: 消息内容（字节数组形式）
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("接收到消息: " + new String(body)); // 打印接收到的消息
            }
        };

        // 开始消费消息
        channel.basicConsume("hello", true, consumer);

        // 等待回调函数执行完毕
        TimeUnit.SECONDS.sleep(5);

        // 7. 释放资源
        // 消费者是一个监听程序，通常情况下不需要主动关闭资源，如果需要手动关闭，可以按照以下顺序
        // channel.close();
        // connection.close();
    }
}

生产者和消费者使用的通道（channel）并不是同一个。当一个新的 RabbitMQ 节点启动时，它会预先声明几个内置交换机，其中默认的内置交换机名称为空字符串（“”）。生产者发送的消息可以根据队列名称直接路由到对应的队列。

如果有一个名为 “hello” 的队列，生产者可以直接将消息发送到 “hello” 队列，而消费者可以从 “hello” 队列中接收消息，而无需关注交换机的存在。这种模式非常适合简单的应用场景，尤其是生产者和消费者之间的一对一通信。