RabbitMQ入门

目录

MQ 相关概念

什么是MQ

MQ 的作用

什么是RabbitMQ

RabbitMQ的安装

安装 erlang 

安装 RabbitMQ

安装 RabbitMQ 管理界面

开放云服务器端口

访问 RabbitMQ 管理界面 

RabbitMQ 的用户角色

RabbitMQ的工作流程

Producer 和 Consumer

Connection 和 Channel

Virrtual host

Queue

Exchange

AMQP

管理界面操作

 添加用户

设置虚拟机操作权限

更新或删除用户

退出当前用户

创建虚拟主机

入门代码

引入依赖

生产者代码实现

创建连接

创建 Channel

声明队列

发送消息

释放资源 

消费者代码实现

消费消息

完整代码

生产者代码

消费者代码

常见问题

IP、端口号不正确或未开放端口号

账号或密码错误

用户对虚拟机没有操作权限

队列不存在

资源释放的顺序

---

MQ 相关概念

什么是MQ

MQ（Message Queue），即 消息队列，因此，它本质上也是一个队列，遵循先进先出（FIFO），队列中存放的内容是 消息（message），消息可以非常简单，如字符串，也可以比较复杂，如内嵌对象

MQ常用于在分布式系统中传递数据，允许不同的应用程序或服务之间进行异步通信

系统之间的调用方式通常有两种：

1. 同步通信

直接调用对方的服务，数据从一端发出后可以立即到达另一端

2. 异步通信

数据从一端出发，先进入到容器中进行临时存储，触发某个条件后，再从容器中发送到另一端。而容器的一个具体实现就是 MQ

因此，MQ的主要工作就是接收并转发消息

MQ 的作用

（1）解耦系统

通过消息队列，生产者和消费者可以不直接联系。生产者发送消息到队列，消费者从队列中读取消息并进行处理。这样，生产者不需要等待消费者处理完毕，就可以继续生产消息。这样各个系统或服务之间通过消息传递数据，而不需要知道对方的实现细节，减少了组件之间的直接依赖

（2）流量控制

通过消息队列，可以暂时缓解系统负载的压力。当系统接收到大量的请求时，生产者可以将请求放入队列中，消费者逐步从队列中处理这些请求，避免瞬间高并发对系统造成的压力

（3）负载均衡

多个消费者可以从同一个队列中获取消息进行处理，从而实现负载均衡。消息队列将消息均匀地分配给多个消费者，提高系统的处理能力和效率

（4）消息分发

当多个系统需要对统一数据做出响应时，可以使用MQ进行消息分发

（5）延迟通知

消息队列可以在特定时间后再处理某些任务，比如定时任务调度、预约任务等

......

而 RabbitMQ 则是 MQ 的一种实现

什么是RabbitMQ

RabbitMQ 是一个开源的消息中间件（Message Broker），实现了消息队列的功能，允许不同的应用程序或服务之间通过消息进行异步通信。RabbitMQ 采用 Erlang 语言实现了 AMQP（Advanced Message Queuing Protocol，高级消息队列协议）标准，提供了可靠的消息传递、队列管理、消息路由等功能

RabbitMQ的安装

此处使用的是 ubuntu 来进行 RabbitMQ 的安装

RabbitMQ 包含在标准的 Ununtu 仓库中，但其包含的版本通常比最新的 RabbitMQ 发行版落后很多，因此，RabbitMQ 团队制作了软件包，可参考官网：Installing on Debian and Ubuntu | RabbitMQ

由于这种方式比较复杂，在这里就使用 Ubuntu 仓库中的版本进行安装

我们先更新软件包：

apt-get update

安装 erlang 

RabbitMQ 需要 Erlang 语言的支持，因此在安装 RabbitMQ 之前需要安装erlang:

apt-get install erlang

 安装完成后，可以通过 erl 命令查看 erlang 版本：

可以使用 halt(). 命令或 ctrl + c 退出：

接下来，我们继续安装 RabbitMQ

安装 RabbitMQ

apt-get install rabbitmq-server

确认是否安装成功：

systemctl status rabbitmq-server

接下来，我们继续安装 RabbitMQ 管理界面

安装 RabbitMQ 管理界面

rabbitmq-plugins enable rabbitmq_management

 

安装完成

接下来，就可以启动服务了（若服务已经启动，则不再需要）：

service rabbitmq-server start

接下来，我们就可以访问 RabbitMQ 的管理界面了

若是使用的是云服务器，则需要先开放端口 15672

开放云服务器端口

找到防火墙或是安全组，点击添加规则：

 成功开放 15672 端口

访问 RabbitMQ 管理界面 

然后就可以通过 IP:port （如：http://49.232.238.62:15672/）访问界面了：

其默认的用户名和密码都为 guest

 

但是却提示：用户只能通过 localhost 登录

这是因为 RabbitMQ 从 3.3.0 开始禁止使用 guest 权限通过除了 localhost 外的访问

那该如何登录呢？

可以添加管理员用户，然后以管理员用户的账号进行登录

添加用户：

rabbitmqctl add_user 账号 密码

添加用户 admin，密码 123456

rabbitmqctl add_user admin 123456

添加成功

接下来，我们为用户添加权限：

 rabbitmqctl set_user_tags 账号 角色名称

为 admin 添加超级管理员角色：

rabbitmqctl set_user_tags admin administrator

添加成功 

接下来，我们就来了解 RabbitMQ 的用户角色

RabbitMQ 的用户角色

 RabbitMQ 的用户角色分为：Administrator、Monitoring、Policymaker、Management、Impersonator 和 None

（1）Administrator（管理员）

管理员角色拥有完全的权限，可以登录管理控制台，进行系统配置、管理用户、管理虚拟主机、队列、交换机、绑定等，具有最高权限

（2）Monitoring（监控）

监控角色只拥有读取权限，可以登录管理控制台，主要用于查看 RabbitMQ 的状态和监控数据

（3）Policymaker（策略制定者）

策略制定者角色主要负责管理 RabbitMQ 的策略配置，可以登录管理控制台，比如定义如何在虚拟主机中应用策略

（4）Management（管理）

管理角色通常具有在管理 UI（Web 界面）中进行操作的权限，仅可登录管理控制台，允许用户管理系统资源

（5）Impersonator（角色代理人）

代理人角色允许一个用户模拟或“冒充”另一个用户执行操作，无法登录管理控制台，用于调试或特殊的操作场景，如运维人员需要模拟其他用户的行为来进行故障排除或审计

（6）None（无权限）

无权限角色（其他用户）表示用户没有任何特殊权限，不被授权访问任何资源，无法登录管理控制台

使用设置的用户名和密码（admin 123456）进行登录：

登录成功： 

 接下来，我们先来学习 RabbitMQ 的工作流程

RabbitMQ的工作流程

RabbitMQ 是一个消息中间件，也是一个生产者消费者模型，负责接收、存储并转发消息

要理解图中的流程，我们需要先来了解一些基本的概念

Producer 和 Consumer

Producer：生产者，是 RabbitMQ 的 客户端，向 RabbitMQ 发送消息

Consumer：消费者，是 RabbitMQ 的 客户端，从 RabbitMQ 接收消息

Broker：就是 RabbitMQ Server，主要用于接收和发送消息

生产者（Producer）创建消息，并将消息发送到 RabbitMQ 中（消息通常是一个带有一定业务逻辑的数据，如 JSON 字符串）消息可以带有一定的标签，RabbitMQ 会根据标签进行路由，将消息发送给符合条件的消费者（Consumer）

消费者（Consumer）连接到 RabbitMQ 服务器就可以消费消息了，在消费的过程中，标签会被丢掉，消费者只会接收到消息，并不会知道消息的生产者是谁

对于 RabbitMQ 而言，一个 RabbitMQ Broker 可以简单看做一个 RabbitMQ 服务节点，或是 RabbitMQ 服务实例，也可以将一个 RabbitMQ Broker 看做一台 RabbitMQ 服务器

Connection 和 Channel

Connection：是客户端与 RabbitMQ 服务器之间的网络连接，通常是通过 TCP 协议建立的。在 RabbitMQ 中，连接是用于承载所有消息通道的基础设施，负责传输客户端和服务器之间的所有数据和控制信息

Channel：是在一个 Connection 上进行消息传递的 "虚拟通道"，一个 Connection 上可以创建多个 Channel，每个 Channel 都是独立的虚拟连接，消息的发送和接收都是基于 Channel 的。通道在 RabbitMQ 中是轻量级的资源，相比 Connection，它的创建和销毁开销较小

Virrtual host

Virtual Host（虚拟主机）：在 RabbitMQ 中，Virtual Host 是一种用于隔离不同应用程序之间消息传递的机制。它允许在同一 RabbitMQ 服务器实例中创建多个逻辑上的隔离空间，每个虚拟主机可以有独立的队列、交换机、绑定、权限等设置

类似于 MySQL 中的 database，一个 MySQL 服务器可以有多个 database

Queue

Queue（队列）：是 RabbitMQ 的内部对象，用于存储消息

多个消费者可以订阅同一个队列

Exchange

Exchange（交换机） ：负责接收生产者发送的消息，并根据特定的规则将这些消息路由到一个或多个 Queue 中

交换机是消息路由的中心，会根据类型和规则来确定如何转发接收到的消息

我们通过一个例子来进一步理解：

寄件人（生产者）的包裹（消息）被送到到物流中心（交换机）后，物流中心（交换机）会根据收件人（消费者）的地址（一定的规则）将快递分发到不同的站点（队列），然后再送到收件人（消费者）手中

其中，进行分发的过程，就是交换机根据类型和规则来将消息转发到队列的过程

此时，我们再来看流程图：

Producer 生产了一条消息 

Producer 需要连接到 RabbitMQ Broker，建立一个连接（Connection），开启一个信道（Channel）

接着，Producer 需要声明一个交换机（Exchange），用于将要发送的消息路由到指定的队列中

Producer 还需要声明一个队列（Queue），用于存放消息

次数，Producer 就可以将消息发送到 RabbitMQ Broker 中了

RabbitMQ Broker 接收消息，并存入对应的队列（Queue）中，若未找到对应队列，则根据生产者的配置，选择丢弃或是退回给生产者

我们还是以送快递为例，进一步进行理解：

RabbitMQ 可以看做物流公司，而 Broker 就像是物流公司的总部，负责协调和管理所有的物流站点，确保快递的安全，高效地运送快递

Virtual host 像是物流公司为不同的客户划分的独立运输中心，每个运输中心都有自己的仓库（Queue），分拣规则（Exchange）和运输路线（Connection 和 Channel），确保不同客户的包裹处理互不干扰，同时可以提供定制化的服务

Queue 就类似于快递站的一个个仓库，用来存放等待派送的包裹，每个仓库都有一个或多个快递员（消费者）负责从仓库中取出包裹并将其派送

Connection 就像是快递员与快递站之间的通信线路，快递员需要通过这个线路来接收和派送快递（消息）

Channel 就像是快递员在运送包裹时的多个并行线路，这样，快递员就可以同时处理多个包裹，如一边派送包裹，一边接收新的包裹

接下来，我们来继续学习 AMQP

AMQP

AMQP (Advanced Message Queuing Protocol) 是一种开放标准的消息传递协议，旨在支持不同平台和编程语言之间的消息通信。AMQP定义了一套确定消息交换的功能，包括交换机（Exchange）、队列（Queue）等。这些组件共同工作，使得生产者能够将消息发送到交换机，然后由队列接收并等待消费者接收。AMQP还定义了一个网络协议，允许客户端应用通过该协议与消息代理和 AMQP 模型进行通信

而 RabbitMQ 是遵从 AMQP 协议的，也就是说，RabbitMQ 是 AMQP 的 Erlang 的实现，且RabbitMQ 还支持 STOWP2、MQTT2 等协议

因此，AMQP 的模型结构与 RabbitMQ 的模型结构是一样的

在了解了基本概念之后，我们来学习如何 RabbitMQ 的管理界面上的相关操作

管理界面操作

我们首先来看 admin，也就是与用户相关的操作

 添加用户

进行添加用户操作：

添加成功：

点击要操作的用户，查看用户详情： 

在用户详情页面，可以进行更新、删除等操作

设置虚拟机操作权限

上述我们添加用户 zhangsan 后，不能操作任何虚拟机

我们可以在用户详情页面对其进行设置：

设置成功

更新或删除用户

在详情页，还可以进行用户的更新（更新密码或权限）和删除操作

退出当前用户

在右上角，可以退出当前用户（log out）

创建虚拟主机

在 admin 页面中，点击右侧的 Virtual Hosts：

添加虚拟主机： 

添加成功：

且此时登录的用户默认可以操作该虚拟主机

若需要对虚拟主机进行设置，可以点击需要设置的虚拟主机，进入详情页

添加 zhangsan 用户

添加成功：

在这里我们先简单学习用户和虚拟机相关的操作，而关于 连接、交换机、队列等操作，我们在后续学习中慢慢进行理解

在学习了上述基础后，我们就来尝试编写代码

入门代码

要编写代码，首先要分析清楚我们要实现的功能：

生产者生产消息，将消息发送给 RabbitMQ，RabbitMQ 将消息发送给消费者

因此，实现的步骤为：

（1）引入依赖

（2）编写生产者代码

（3）编写消费者代码

引入依赖

        <!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client --><dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.20.0</version></dependency>

生产者代码实现

生产者要向 RabbitMQ 发送消息，首先需要与 RabbitMQ 建立连接

创建连接

而建立连接需要的信息有：

（1）IP

（2）端口号

（3）账号

（4）密码

（5）需要使用的虚拟主机

RabbitMQ 默认用于与客户端连接的 TCP 端口号是 5672，因此，需要提前进行开放：

为了方便演示，我们直接在 main 方法中编写代码： 

public class Producer {public static void main(String[] args) throws IOException, TimeoutException {// 1. 创建连接工厂ConnectionFactory factory = new ConnectionFactory();// 2. 设置参数factory.setHost("49.232.238.62"); // ip 的默认值为 localhostfactory.setPort(5672); // 默认值为 5672factory.setVirtualHost("test01"); // 虚拟主机，默认值为 /// 账号factory.setUsername("admin"); // 用户名，默认为 guestfactory.setPassword("123456"); // 密码，默认为 guest// 3. 创建连接 ConnectionConnection connection = factory.newConnection(); // 需要处理异常，在此处直接进行抛出}
}

 连接建立好后，我们接着来创建 Channel

创建 Channel

        // 4. 创建 ChannelChannel channel = connection.createChannel();

声明队列

需要使用 queueDeclare 方法来声明队列

Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) throws IOException;

我们来看这个方法中的参数和返回值：

参数： 

queue：队列的名称

durable：队列是否持久化，若设置为 true，则持久化的队列会存盘，服务器重启后，消息不会丢失

exclusive：是否独占，若设置为 true，则只有一个消费者能监听队列

autoDelete：是否自动删除，当没有消费者（Consumer）时，是否自动删除队列

arguments：用于设置队列的属性

返回值：

Queue.DeclareOk：声明确认方法，用于标识队列已成功声明

声明一个名为 simple.test 的持久化队列：

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

simple.test 队列不被一个消费者独占，不会自动删除，队列的属性不进行设置

声明队列后，若不存在 simple.test 队列，则会自动创建；若存在，则不进行创建，使用已有的

接下来，我们就可以向 RabbitMQ 发送消息了

发送消息

在前面我们了解到，生产者发送的消息，由 Exchange（交换机）接收，并根据特定的规则将这些消息路由到一个或多个 Queue 中

当一个新的 RabbitMQ 节点启动时，会预声明（declare）几个内置的交换机，内置交换机名称是空字符串（""），生产者发送的消息会根据队列名称直接路由到对应的队列

例如，有一个名为 "simple.test" 的队列，当我们使用内置的交换机时，可以看做生产者直接发送消息到  "simple.test" 队列中，消费者可以直接从  "simple.test" 队列中接收消息，而不需要关心交换机的存在。这种模式适合非常简单的应用场景，生产者和消费者之间的通信是一对一的

在这里，我们就使用内置的交换机来将生产者的消息存储到  "simple.test" 队列中

通过 basicPublish 方法来进行发送

我们来看 basicPublish 方法：

void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException;

exchange：交换机名称，若使用内置的交换机，默认为 ""

routingKey：路由键，当使用内置交换机时，routingKey 为队列名称

props：配置相关信息

body：要发送的消息

        // 6. 通过 channel 发送消息到队列中String message = "test...";channel.basicPublish("", "simple.test", null, message.getBytes());System.out.println("消息：" + message + " 发送成功");

释放资源 

当我们发送完消息后，不要忘记释放资源

        // 7. 释放资源channel.close();connection.close();

当 Connection 关闭时，Channel 也会自动关闭，因此，Channel 可以不用关闭，但是显示关闭 Channel 是一个良好的习惯

生产者的代码编写完成后，我们就可以运行代码，并观察结果了

我们先打开 RabbitMQ 管理界面 中的 Queues and Streams 界面：

运行代码，再次观察：

此时，创建了 simple.test 队列，且队列中有一条消息

我们查看 simple.test 的详细信息：

点击 Get messages：  

点击 Get Message(s) 获取队列中的一条消息，可以看到， 生产者的消息成功发送到 simple.test 队列中

此时，我们注释掉释放资源的代码，再次运行程序：

 就可以在 Connections 和 Channels 中查看到相关信息了：

接下来，我们继续编写消费者的代码

消费者代码实现

消费者要从 RabbitMQ 中获取消息，也需要先与其建立连接，并创建 Channel，同时声明队列 Queue

public class Consumer {public static void main(String[] args) throws IOException, TimeoutException {// 1. 创建连接工厂ConnectionFactory factory = new ConnectionFactory();// 2. 设置参数factory.setHost("49.232.238.62"); // ip 的默认值为 localhostfactory.setPort(5672); // 默认值为 5672factory.setVirtualHost("test01"); // 虚拟主机，默认值为 /// 账号factory.setUsername("admin"); // 用户名，默认为 guestfactory.setPassword("123456"); // 密码，默认为 guest// 3. 创建连接 ConnectionConnection connection = factory.newConnection(); // 需要处理异常，在此处直接抛出，并不进行处理// 4. 创建 ChannelChannel channel = connection.createChannel();// 5. 声明队列channel.queueDeclare("simple.test", true, false, false, null);}
}

也可以不声明队列，但是若队列不存在，就会抛出异常，因此，我们在编写消费者代码时也声明队列 simple.test

接下来，我们就可以消费队列中的消息了

消费消息

使用 basicConsume 方法来消费消息：

String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;

queue：队列名称

autoAck：是否自动确认，消费者收到消息后，自动向 MQ 确认已经收到消息

callback：回调对象

我们重点来看 callback 这个回调对象，其类型为 Consumer：

Consumer 是一个接口，用于定义消息消费者的行为，当我们需要从 RabbitMQ 接收消息时，需要提供一个实现了 Consumer 接口的对象

而 DefaultConsumer 是 RabbitMQ 提供的一个默认消费者，实现类 Consumer 接口

其中，handleDelivery 方法是其核心方法

当从队列中接收到消息后，会自动调用该方法，在方法中，我们可以实现如何处理接收到的消息，如打印消息内容、将消息存储到数据库等

其中的参数有：

consumerTag：消费者标签，通常是消费者的订阅队列时指定

envelope：包含消息的封包信息，如队列名称、交换机名称等

properties：配置信息

body：消息内容

我们定义 DefaultConsumer 对象，并重写 handleDelivery 方法，在方法中实现消息的处理逻辑：

        // 6. 消费消息DefaultConsumer consumer = new DefaultConsumer(channel) {// 回调方法，当接收到消息后，自动执行该方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("成功接收到消息: " + new String(body));}};

在这里，我们就简单的打印接收到的消息

再使用  basicConsume 方法消费消息：

        channel.basicConsume("simple.test", true, consumer);

消费 simple.test 队列中的消息，当消费者接收到消息后，自动向 RabbitMQ 进行确认

最后，释放资源：

        // 7. 释放资源channel.close();connection.close();

消费者相当于是一个监听程序，需要监听队列中是否有消息需要消费，因此，在大多数情况下，不需要关闭资源

 我们运行消费者代码，并观察结果：

由于刚才生产者代码运行了两次，因此向队列中发送了两条消息，而这两条消息都成功被消费者接收

若我们不释放消费者的资源，则可以看到响应的 Connection 和 Channel

 在 Queues and Streams 页面，点击右侧的 +/-： 

点击添加消费者数量： 

可以看到此时 simple.test 队列有一个消费者： 

完整代码

以下是生产者和消费者的完整代码：

生产者代码

public class Producer {public static void main(String[] args) throws IOException, TimeoutException {// 1. 创建连接工厂ConnectionFactory factory = new ConnectionFactory();// 2. 设置参数factory.setHost("49.232.238.62"); // ip 的默认值为 localhostfactory.setPort(5672); // 默认值为 5672factory.setVirtualHost("test01"); // 虚拟主机，默认值为 /// 账号factory.setUsername("admin"); // 用户名，默认为 guestfactory.setPassword("123456"); // 密码，默认为 guest// 3. 创建连接 ConnectionConnection connection = factory.newConnection(); // 需要处理异常，在此处直接抛出，并不进行处理// 4. 创建 ChannelChannel channel = connection.createChannel();// 5. 声明队列channel.queueDeclare("simple.test", true, false, false, null);// 6. 通过 channel 发送消息到队列中String message = "test...";channel.basicPublish("", "simple.test", null, message.getBytes());System.out.println("消息：" + message + " 发送成功");// 7. 释放资源channel.close();connection.close();}
}

消费者代码

public class Consumer {public static void main(String[] args) throws IOException, TimeoutException {// 1. 创建连接工厂ConnectionFactory factory = new ConnectionFactory();// 2. 设置参数factory.setHost("49.232.238.62"); // ip 的默认值为 localhostfactory.setPort(5672); // 默认值为 5672factory.setVirtualHost("test01"); // 虚拟主机，默认值为 /// 账号factory.setUsername("admin"); // 用户名，默认为 guestfactory.setPassword("123456"); // 密码，默认为 guest// 3. 创建连接 ConnectionConnection connection = factory.newConnection(); // 需要处理异常，在此处直接抛出，并不进行处理// 4. 创建 ChannelChannel channel = connection.createChannel();// 5. 声明队列channel.queueDeclare("simple.test", true, false, false, null);// 6. 消费消息DefaultConsumer consumer = new DefaultConsumer(channel) {// 回调方法，当接收到消息后，自动执行该方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("成功接收到消息: " + new String(body));}};channel.basicConsume("simple.test", true, consumer);// 7. 释放资源channel.close();connection.close();}
}

常见问题

IP、端口号不正确或未开放端口号

若连接的 IP 或是端口号不正确，又或是未开放端口号，则会连接不上，等待一段时间，程序就会抛出异常：

连接超时，没有进一步的信息

此时，就需要检查 IP 和 端口号是否正确，以及端口号是否已经开放

账号或密码错误

当账号或密码错误时，会抛出异常：

此时就需要检查用户名和密码是否正确

用户对虚拟机没有操作权限

进入 admin 详情页，删除 admin 对 test01 虚拟主机的操作权限：

此时再次进行访问：

当前用户对 test01 虚拟主机没有操作权限

此时我们就需要检查当前用户的操作权限

队列不存在

找到 simple.test 队列，进入详情页：

将其删除： 

并注释掉队列声明的代码，此时再次运行程序：

在虚拟主机 test01 上不存在 simple.test 队列

因此，我们最好每次都将声明队列加上，防止队列不存在时抛出异常，影响后续程序的执行

资源释放的顺序

在关闭资源时，一定是先关闭 Channel，再关闭 Connectiion

若先关闭 Connection，再关闭 Channel，就会出现问题：

这是因为当 Connection 关闭时，Channel 也会自动关闭，而此时再尝试关闭 Channel，自然会报错