在这里插入图片描述

前言

前面我们介绍了什么是 RabbitMQ,以及 RabbitMQ 的作用和优势,那么这篇文章我将为大家介绍如何安装 RabbitMQ,以及如何使用 RabbitMQ。

安装RabbitMQ

我这里选择的安装环境是在 Linux 的 Ubuntu 发行版本中安装的,大家的平台如果不一样的话,大家可以去官网看看如何安装。RabbitMQ各个系统的安装

前面我们说了 RabbitMq 是由 Erlang 语言实现的,所以要想使用 RabbitMQ,我们首先就需要安装 Erlang。

在安装之前,我们先使用 apt-get update 来更新 apt 库:

在这里插入图片描述
然后使用 apt-get install erlang 下载 Erlang:

在这里插入图片描述
下载完成之后 Ubuntu 会提示我们需要启动什么服务,我们按 ESC 退出,然后查看 Erlang 是否安装成功。在 Linux 命令行中输入 erl:

在这里插入图片描述
如果显示出来 Erlang 的版本就说明安装成功。当进入 Erlang 之后,我们如何退出呢?可以通过 halt(). 命令退出,也可以直接使用 CTRL + c 退出。

Erlang 安装成功之后,我们再来安装 RabbitMQ,安装 RabbitMQ 和安装 Erlang 是类似的,通过 apt-get install rabbitmq-server

在这里插入图片描述

安装完成之后,我们通过 systemctl status rabbitmq-server 来查看 rabbit-server 的状态:

在这里插入图片描述
如果显示 active(running)则表示安装成功。

通过 apt-get install rabbitmq-server 是安装了 rabbitmq 的服务,那么我们还需要安装的东西就是 RabbitMQ 的管理页面,rabbitmq-plugins enable rabbitmq_management

在这里插入图片描述

安装完成 RabbitMQ 的管理页面之后,我们就可以通过 service rabbitmq-server statrt 启动 RabbitMQ 的服务,然后访问这个管理页面了:

启动 RabbitMQ 之后,我们就通过 IP+Port 的方式来访问 rabbitmq 的管理页面,rabbitmq 的管理页面默认使用的端口号是 15671。

在这里插入图片描述

与 RabbitMQ 相关的端口号有三个 5672、15672和25672,5671 是客户端和 RabbitMQ 连接时所使用的端口号,15672 就是访问管理页面所使用的端口号,25672 则是我们搭建集群的时候所需要使用到的端口号。

在这里插入图片描述
通过 IP + Port 的方式进入管理页面之后,这里提示我们输入用户名和密码,但是因为我们第一次进入这个管理页面,所以也没有创建账号,虽然 rabbitmq 为我们提供了几个默认的账号,但是现在已经无法登录进去了,所以需要我们创建账户之后才能登录。那么我们如何创建账户呢?

我们在 Linux 命令行中输入 rabbitmqctl add_user ${账号} ${密码},就可以创建一个帐号了:

在这里插入图片描述

创建完成之后,我们直接登陆的话还是不能登陆进去:

在这里插入图片描述
这是为什么呢?因为这时候我们创建的用户还不具有权限,所以我们接下来需要给当前用户进行赋权操作 rabbitmqctl set_user_tags ${账号} ${角色名称},这里的角色名称有下面几种可以选择:

  1. Administrator 超级管理员,可登录登录管理控制台(启用management plugin的情况下),可查看所有的信息,并且可以对用户,策略(policy)进行操作
  2. Monitoring 监控者,可登录登录管理控制台(启用management plugin的情况下),同时可以查看 rabbitmq 节点的相关信息(进程数,内存使用情况,磁盘使用情况等)
  3. Policymaker 策略制定者,可登录登录管理控制台(启用management plugin的情况下),同时可以对 policy 进行管理,但无法查看节点的相关信息
  4. Management 普通管理者,仅可登录管理控制台(启用management plugin的情况下),无法看到节点信息,也无法对策略进行管理
  5. Impersonator 模拟者,无法登录管理控制台
  6. None 其他用户,无法登录管理控制台,通常就是普通的生产者和消费者

我们这里就给用户设置为 administrator 角色 rabbitmqctl set_user_tags admin administrator

在这里插入图片描述

为账号赋权之后,我们再登录:

在这里插入图片描述

RabbitMQ核心概念

安装完成 RabbitMQ 并且进入到管理页面之后,我们可以看到上面的导航栏有六个部分,那么这六个部分分别代表什么?

Producer 和 Consumer

Producer:生产者,是 RabbitMQ Server 的客户端,向 RabbitMQ 发送消息
Consumer:消费者,也是 RabbitMQ Server 的客户端,从 RabbitMQ 接收消息
Broker:就是 RabbitMQ,主要是接收和发送消息

  • 当生产者和 RabbitMQ 建立连接之后,生产者会创建消息,然后发布到 RabbitMQ 中。在实际应用中,消息通常是一个带有一定业务逻辑结构的数据,比如 JSON 字符串。消息可以带有一定的标签,RabbitMQ 会根据标签进行路由,把消息发送给感兴趣的消费者
  • 消费者连接到 RabbitMQ 服务器就可以消费消息了,消费的过程中,标签会被丢掉。消费者只会收到消息,并不知道消息的生产者是谁,当然消费者也不需要知道
  • 对于 RabbitMQ 来说,一个 RabbitMQ Broker 可以简单的看作一个 RabbitMQ 服务节点,或者 RabbitMQ 服务实例,大多数情况下也可以将一个 RabbitMQ Broker 看作一台 RabbitMQ 服务器

Connection 和 Channel

Connection:连接,是客户端和 RabbitMQ 服务器之间的一个 TCP 连接,这个连接是建立消息传递的基础,它负责传输客户端和服务器之间的所有数据和控制信息

Channel: 通道,信道,Channel 是在 Connection 之上的一个抽象层。在 RabbitMQ 中,一个 TCP 连接可以有多个 Channel,每个 Channel 都是独立的虚拟连接。消息的发送和接收都是基于 Channel 的

通道的主要作用就是将消息的读写操作复用到一个 TCP 连接中,这样可以减少创建和销毁的开销,提高性能。我们可以大致的将 Connection 和 Channel 之间的关系看成是进程和线程之间的关系。

在这里插入图片描述

Virtual host

Virtual host:虚拟主机,这是一个虚拟的概念,它为消息队列提供了一种逻辑上的隔离机制,对于一个 RabbitMQ 而言,一个 BrokerServer 上可以存在多个 Virtual Host。当多个不同的用户使用同一个 RabbitMQ Server 提供的服务时,可以虚拟划分出多个 vhost,每个用户在自己的 vhost 创建 exchange/queue 等。Virtual Host 就类似我们 MySQL 中的 database,一个主机上的 MySQL 可以为多个项目提供服务,每个项目使用不同的 database。

Queue

Queue 队列,RabbitMQ 的内部对象,用于存储消息,也就不过多介绍了。

Exchange

Exchange 交换机,生产者生产消息到达 broker 的第一站,它负责接收生产者发送的消息,并根据特定的规则把这些消息路由到一个或者多个 Queue 中。

在这里插入图片描述

RabbitMQ 工作流程

在这里插入图片描述
RabbitMQ 是一个消息中间件,也是一个消费者生产者模型,它负责接收、存储并转发消息。消息传递的过程类似邮局,当你要发送一个邮件时,你会把你的邮件放到邮局,邮局接收到邮件,并通过邮递员送到收件人的手中。在上面的图中,Producer 就相当于发件人,Consumer 就相当于收件人,RabbitMQ 就类似于邮局。

生产者发送消息的流程

  1. 建立连接与开启信道:
  • 生产者首先与RabbitMQ建立TCP连接(Connection),并基于这个连接开启一个或多个信道(Channel)。信道是建立在连接之上的虚拟连接,RabbitMQ处理的每一条AMQP指令都是通过信道完成的。这种方式类似于NIO的做法,复用TCP连接,减少性能开销,便于管理。
  1. 声明交换器(Exchange):
  • 生产者声明一个交换器,并设置相关属性,如交换器类型(如direct、topic、fanout等)、是否持久化等。交换器是RabbitMQ中用于接收生产者发送的消息,并根据路由键(routingKey)将消息路由到一个或多个队列中的组件。
  1. 声明队列:
  • 生产者声明一个队列,并设置相关属性,如是否排他、是否持久化、是否自动删除等。队列是RabbitMQ中用于存储消息的容器。
  1. 绑定交换器和队列:
  • 生产者通过绑定键(bindingKey)将交换器和队列绑定起来。这样,当交换器接收到消息时,就可以根据路由键和绑定键的匹配规则,将消息路由到相应的队列中。
  1. 发送消息:
  • 生产者发送消息至RabbitMQ Broker,消息中包含路由键、交换器等信息。
  1. 消息路由与存储:
  • 相应的交换器根据接收到的路由键查找相匹配的队列。如果找到,则将从生产者发送过来的消息存入相应的队列中;如果没有找到,则根据生产者配置的属性选择丢弃还是回退给生产者。
  1. 关闭信道与连接:
  • 生产者完成消息发送后,关闭信道和连接。

消费者接收消息的流程

  1. 建立连接与开启信道:
  • 消费者与RabbitMQ建立TCP连接,并开启一个或多个信道。
  1. 请求消费消息:
  • 消费者向RabbitMQ请求消费相应队列中的消息,可能会设置相应的回调函数以及做一些准备工作。
  1. 接收消息:
  • 等待RabbitMQ回应并投递相应队列中的消息,消费者接收消息。
  1. 消息确认:
  • 消费者确认(ack)接收到的消息。一旦消息被确认,RabbitMQ就会从队列中删除该消息。
  1. 关闭信道与连接:
  • 消费者完成消息消费后,关闭信道和连接。

AMQP

前面反复的提到 AMQP,那么什么是 AMQP 呢?

AMQP(Advanced Message Queuing Protocol),即高级消息队列协议,是一个提供统一消息服务的应用层标准协议,是应用层协议的一个开放标准。AMQP专为面向消息的中间件设计,基于此协议的客户端与消息中间件传递消息,并不受客户端/中间件不同产品、不同开发语言等条件的限制。

AMQP的工作流程主要包括以下几个步骤:

  1. 生产者发送消息:生产者将消息发送到指定的交换器(Exchange),并指定路由键(Routing Key)。
  2. 交换器路由消息:交换器根据路由键和绑定键(Binding Key)的匹配规则,将消息路由到相应的队列(Queue)中。
  3. 消费者接收消息:消费者订阅指定的队列,并从队列中接收消息进行处理。
  4. 消息确认:消费者处理完消息后,会向RabbitMQ发送确认消息,RabbitMQ收到确认后,会将该消息从队列中删除。

在这里插入图片描述

Web页面操作

但知道了 RabbitMQ 的核心概念和工作流程之后,我们来看看 RabbitMQ 的管理页面如何使用吧。

在这里插入图片描述
这里的 Connections、Channels、Exchanges和Queues因为我们还没有客户端连接 RabbitMQ 所以此时还没有什么信息,后面有客户端连接 RabbitMQ 的时候再来看看这里面的内容。

然后就是这个 Admin,账号管理:

在这里插入图片描述
在这里插入图片描述
我们可以点击对应的账号,对其权限、可操作的 Virtual Host 等进行设置:

在这里插入图片描述
也可以对绑定的交换机进行设置:

在这里插入图片描述
修改账号的密码和权限:

在这里插入图片描述
删除账号:

在这里插入图片描述
也可以在 Admin 页面添加账号:

在这里插入图片描述

我们也可以点击右边的 Virtual Host 对虚拟主机进行设置:

在这里插入图片描述
在这里我们可以看到所有的虚拟主机,以及每一个主机中哪些用户对其具有权限:

在这里插入图片描述
当然也可以创建一个新的虚拟主机:

在这里插入图片描述
在这里插入图片描述
创建完成之后,默认是哪个用户创建的这个虚拟主机,这个用户就对这个虚拟主机具有权限。

在这里插入图片描述
我们可以点击某个特定的虚拟主机对齐进行设置:

在这里插入图片描述
可以删除或者增加某个用户对这个虚拟主机的权限:

在这里插入图片描述

RabbitMQ快速入门

在知道了上面的知识了之后,我们来看看 Java 中如何使用 RabbitMQ。

1. 引入依赖

RabbitMQ 属于第三方库,不在 java.lang 包下,所以需要我们将 RabbitMQ 的依赖引入才能使用 RabbitMQ。

在这里插入图片描述
在这里插入图片描述
将复制的坐标粘贴到 pom.xml 文件中:

在这里插入图片描述
导入 rabbitmq 依赖之后,我们来分别编写生产者代码和消费者代码。

2. 编写生产者代码

在这里插入图片描述
根据这个 rabbitmq 的工作流程图,我们来看看生产者需要做什么。

1.首先生产者需要先和 RabbitMQ Broker 建立连接

  1. 创建连接工厂:

在这里插入图片描述

2)设置参数,需要的参数有 IP、Port、Virtual Host、Username、Password:

public class ProducerDemo {
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory(); //创建连接工厂
        factory.setHost("x.x.x.x"); //设置IP
        factory.setPort(5672); //前面说了,客户端与RabbitMQ Server连接使用的端口默认是5672
        factory.setVirtualHost("test"); //Virtual Host
        factory.setUsername("admin"); //用户名
        factory.setPassword("****"); //密码
        //创建连接Connection
        Connection connection = factory.newConnection();
    }
}

2.开启信道

Channel channel = connection.createChannel();

3.声明交换机,因为 rabbitmq 默认提供了几个交换机,所以这里我们就使用默认的交换机。

4.声明队列

channel.queueDeclare() 的参数解释:

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

  • String queue:这是你想要声明(或创建)的队列的名称。如果队列已经存在,并且参数与现有队列的设置兼容,那么这个方法会简单地确认队列的存在。如果队列不存在,则根据提供的参数创建新队列。
  • boolean durable:此参数指定队列是否应该持久化。如果设置为true,那么队列会在RabbitMQ重启后仍然存在。如果设置为false,那么队列是临时的,当RabbitMQ重启后,该队列将不再存在(除非它已经被声明为持久化)。
  • boolean exclusive:此参数指定队列是否应该是排他的。如果设置为true,则队列只能由声明它的连接(connection)使用,并且该队列会在连接关闭时自动删除。设置为false时,队列可以被多个连接共享。
  • boolean autoDelete:此参数指定队列是否应该自动删除。如果设置为true,那么当没有任何消费者(consumer)连接到队列时,队列会自动删除。这通常用于临时队列。如果设置为false,则队列不会自动删除。
  • Map<String, Object> arguments:这是一个可选参数,允许你指定一系列队列的额外参数(也称为“arguments”或“x-arguments”)。这些参数可以是RabbitMQ特有的,用于控制队列的行为,比如设置队列的消息最大长度、死信交换器(dead-letter exchange)等。
channel.queueDeclare("hello",true,false,false,null);

5.发送消息

void basicPublish(String exchange, String routingKey, AMQP.BasicProperties props, byte[] body) throws IOException;
  • String exchange:指定消息要发送到的交换机的名称。交换机是 RabbitMQ 中的核心概念,它接收生产者发送的消息,并根据路由键(routing key)将消息路由到一个或多个队列。
  • String routingKey:路由键是一个字符串,用于将消息路由到交换机所连接的队列。交换机使用路由键来确定消息应该被发送到哪个队列(或多个队列)。如果交换机类型为 direct,则路由键必须完全匹配队列的绑定键(binding key)。
  • AMQP.BasicProperties props:这是一个可选参数,用于指定消息的属性,如内容类型(content type)、内容编码(content encoding)、消息优先级(priority)、消息过期时间(expiration)、消息ID(message-id)、用户ID(user-id)、应用程序ID(app-id)等。这些属性可以在消息被消费时由消费者使用,以决定如何处理消息。
  • byte[] body:这是消息的实际内容,以字节数组的形式提供。这个字节数组可以包含任何类型的数据,但通常包含的是序列化后的对象或纯文本数据。
for (int i = 0; i < 10; i++) {
    String msg = "hello rabbitmq" + i;
    channel.basicPublish("","hello",null,msg.getBytes()); //因为使用的是默认的交换机,所以交换机这个参数的值就是空字符串
    System.out.println("消息发送成功");
}

6.释放资源

// 释放资源,先释放信道再释放Connection
channel.close();
connection.close();

当这里我们释放资源了之后,管理页面有些内容我们就看不到了,所以,我们先不释放资源。

编写完成生产者代码之后,我们运行这段代码,然后查看管理页面中的变化:

首先是 Connection:

在这里插入图片描述
Chanel:

在这里插入图片描述
Exchange:

在这里插入图片描述

Queue:

在这里插入图片描述
点进去队列,我们可以看到队列的详细信息:

在这里插入图片描述
在这里插入图片描述
我们也可以在这个界面中生产消息:

在这里插入图片描述
得到队列中的 n 条消息:

在这里插入图片描述

3. 编写消费者代码

1.建立连接

消费者跟生产者一样,首先还是要和 RabbitMQ Broker 建立连接:

public class ConsumerDemo {
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory(); //创建连接工厂
        factory.setHost("x.x.x.x"); //设置IP
        factory.setPort(5672); //前面说了,客户端与RabbitMQ Server连接使用的端口默认是5672
        factory.setVirtualHost("test"); //Virtual Host
        factory.setUsername("admin"); //用户名
        factory.setPassword("***"); //密码
        //建立连接Connection
        Connection connection = factory.newConnection();
    }
}

2.创建Channel

connection.createChannel();

3.声明交换机,这里还是使用默认的交换机

4.声明队列

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

5.消费消息

basicConsume(String queue, boolean autoAck, Consumer callback);
  • String queue: 这个参数指定了消费者将从哪个队列中接收消息。队列名是一个字符串,它标识了 RabbitMQ 服务器上的一个特定队列。在调用此方法之前,该队列应该已经被创建,或者 RabbitMQ 服务器配置为在首次尝试访问时自动创建队列。
  • boolean autoAck: 这个参数是一个布尔值,用于指定是否自动确认(acknowledge)接收到的消息。
    如果设置为 true,则 RabbitMQ 会认为一旦消息被消费者接收(即调用 basicConsume 后,消息被传递给回调函数),该消息就已经被成功处理,并会立即从队列中删除。这种方式简单但风险较高,因为如果消费者在处理消息时发生异常或崩溃,那么这些消息就会丢失,因为它们已经被确认并从队列中移除了。
    如果设置为 false,则消费者需要显式地通过调用 basicAck 方法来确认消息已被成功处理。这种方式提供了更高的可靠性,因为只有在消费者明确确认后,消息才会从队列中删除。如果消费者在处理消息时失败或崩溃,RabbitMQ 会重新将消息发送给其他消费者(如果配置了消息重试或死信队列等策略)。
  • Consumer callback: 这是一个回调函数,当有新消息到达指定队列时,RabbitMQ 会调用这个函数。回调函数的实现需要由开发者提供,它定义了当接收到消息时应该执行的操作。这个回调函数通常接受一个 Delivery 对象作为参数,该对象包含了消息的内容、消息的属性(如消息ID、优先级等)以及用于确认消息(如果 autoAck 为 false)的通道(Channel)等信息。
Consumer consumer = new DefaultConsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        System.out.println("接收到消息" + new String(body));
    }
};
channel.basicConsume("hello",false,consumer);

运行 Consumer:

在这里插入图片描述
查看管理界面变化:

多出了一个连接Connection
在这里插入图片描述
Channel 也是多出了一个

在这里插入图片描述
但是 Exchange 交换机没有变化,因为交换机是只有生产者到达 Broker 的时候才会使用到,从 Broker 到消费者之间不会使用到 Exchange:

在这里插入图片描述
因为我这里没有设置自动应答,也没有手动应答,所以所有的消费的消息都显示的是 Unacked:

在这里插入图片描述

点赞(0) 打赏

评论列表 共有 0 条评论

暂无评论

微信公众账号

微信扫一扫加关注

发表
评论
返回
顶部