文章目录
前言
前面我们介绍了什么是 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 ${账号} ${角色名称}
,这里的角色名称有下面几种可以选择:
- Administrator 超级管理员,可登录登录管理控制台(启用management plugin的情况下),可查看所有的信息,并且可以对用户,策略(policy)进行操作
- Monitoring 监控者,可登录登录管理控制台(启用management plugin的情况下),同时可以查看 rabbitmq 节点的相关信息(进程数,内存使用情况,磁盘使用情况等)
- Policymaker 策略制定者,可登录登录管理控制台(启用management plugin的情况下),同时可以对 policy 进行管理,但无法查看节点的相关信息
- Management 普通管理者,仅可登录管理控制台(启用management plugin的情况下),无法看到节点信息,也无法对策略进行管理
- Impersonator 模拟者,无法登录管理控制台
- 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 就类似于邮局。
生产者发送消息的流程
- 建立连接与开启信道:
- 生产者首先与RabbitMQ建立TCP连接(Connection),并基于这个连接开启一个或多个信道(Channel)。信道是建立在连接之上的虚拟连接,RabbitMQ处理的每一条AMQP指令都是通过信道完成的。这种方式类似于NIO的做法,复用TCP连接,减少性能开销,便于管理。
- 声明交换器(Exchange):
- 生产者声明一个交换器,并设置相关属性,如交换器类型(如direct、topic、fanout等)、是否持久化等。交换器是RabbitMQ中用于接收生产者发送的消息,并根据路由键(routingKey)将消息路由到一个或多个队列中的组件。
- 声明队列:
- 生产者声明一个队列,并设置相关属性,如是否排他、是否持久化、是否自动删除等。队列是RabbitMQ中用于存储消息的容器。
- 绑定交换器和队列:
- 生产者通过绑定键(bindingKey)将交换器和队列绑定起来。这样,当交换器接收到消息时,就可以根据路由键和绑定键的匹配规则,将消息路由到相应的队列中。
- 发送消息:
- 生产者发送消息至RabbitMQ Broker,消息中包含路由键、交换器等信息。
- 消息路由与存储:
- 相应的交换器根据接收到的路由键查找相匹配的队列。如果找到,则将从生产者发送过来的消息存入相应的队列中;如果没有找到,则根据生产者配置的属性选择丢弃还是回退给生产者。
- 关闭信道与连接:
- 生产者完成消息发送后,关闭信道和连接。
消费者接收消息的流程
- 建立连接与开启信道:
- 消费者与RabbitMQ建立TCP连接,并开启一个或多个信道。
- 请求消费消息:
- 消费者向RabbitMQ请求消费相应队列中的消息,可能会设置相应的回调函数以及做一些准备工作。
- 接收消息:
- 等待RabbitMQ回应并投递相应队列中的消息,消费者接收消息。
- 消息确认:
- 消费者确认(ack)接收到的消息。一旦消息被确认,RabbitMQ就会从队列中删除该消息。
- 关闭信道与连接:
- 消费者完成消息消费后,关闭信道和连接。
AMQP
前面反复的提到 AMQP,那么什么是 AMQP 呢?
AMQP(Advanced Message Queuing Protocol),即高级消息队列协议,是一个提供统一消息服务的应用层标准协议,是应用层协议的一个开放标准。AMQP专为面向消息的中间件设计,基于此协议的客户端与消息中间件传递消息,并不受客户端/中间件不同产品、不同开发语言等条件的限制。
AMQP的工作流程主要包括以下几个步骤:
- 生产者发送消息:生产者将消息发送到指定的交换器(Exchange),并指定路由键(Routing Key)。
- 交换器路由消息:交换器根据路由键和绑定键(Binding Key)的匹配规则,将消息路由到相应的队列(Queue)中。
- 消费者接收消息:消费者订阅指定的队列,并从队列中接收消息进行处理。
- 消息确认:消费者处理完消息后,会向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 建立连接
- 创建连接工厂:
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:
本站资源均来自互联网,仅供研究学习,禁止违法使用和商用,产生法律纠纷本站概不负责!如果侵犯了您的权益请与我们联系!
转载请注明出处: 免费源码网-免费的源码资源网站 » RabbitMQ快速上手
发表评论 取消回复