【RabbitMQ】 RabbitMQ快速上手_rabbitmq详细教程

技术文档

文章目录

一、RabbitMQ 核心概念
- 1.1 Producer和Consumer
- 2.2 Connection和Channel
- 2.3 Virtual host
- 2.4 Queue
- 2.5 Exchange
- 2.6 RabbitMQ工作流程
二、AMQP协议
三、web界面操作
- 4.1 用户相关操作
- 4.2 虚拟主机相关操作
四、RabbitMQ快速入门
- 4.1 引入依赖
- 4.2 编写生产者代码
- - 4.2.1 创建连接
  - 4.2.2 创建Channel
  - 4.2.3 声明⼀个队列Queue
  - 4.2.4 发送消息
  - 4.2.5 释放资源
  - 4.2.6 观察结果
- 4.3 编写消费者代码
- - 4.3.1 消费当前队列
  - 4.3.2 释放资源
  - 4.3.3 观察结果
- 4.4 源码
结语

一、RabbitMQ 核心概念

我们接下来将学习如何去使用RabbitMQ。
下图是RabbitMQ的管理页面：
【RabbitMQ】 RabbitMQ快速上手_rabbitmq详细教程
界面上的导航栏共分6部分, 这6部分分别是什么意思呢, 我们先看看RabbitMQ的工作流程：

【RabbitMQ】 RabbitMQ快速上手_rabbitmq详细教程

RabbitMQ是⼀个消息中间件, 也是⼀个生产者消费者模型. 它负责接收, 存储并转发消息.

消息传递的过程类似邮局. 当你要发送⼀个邮件时，你把你的邮件放到邮局，邮局接收邮件, 并通过邮递员送到收件人的手上

按照这个逻辑, Producer 就类似邮件发件人. Consumer 就是收件人, RabbitMQ就类似于邮局

1.1 Producer和Consumer

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

⽣产者(Producer)创建消息, 然后发布到RabbitMQ中. 在实际应⽤中, 消息通常是⼀个带有⼀定业务逻辑结构的数据, ⽐如JSON字符串. 消息可以带有⼀定的标签, RabbitMQ会根据标签进行路由, 把消息发送给感兴趣的消费者(Consumer).

消费者连接到RabbitMQ服务器, 就可以消费消息了, 消费的过程中, 标签会被丢掉. 消费者只会收到消息, 并不知道消息的⽣产者是谁, 当然消费者也不需要知道.

对于RabbitMQ来说,⼀个RabbitMQ Broker可以简单地看作⼀个RabbitMQ服务节点, 或者RabbitMQ服务实例. ⼤多数情况下也可以将⼀个RabbitMQ Broker看作⼀台RabbitMQ服务器

2.2 Connection和Channel

Connection: 连接. 是客户端和RabbitMQ服务器之间的⼀个TCP连接. 这个连接是建⽴消息传递的基础, 它负责传输客⼾端和服务器之间的所有数据和控制信息.
Channel: 通道, 信道. Channel是在Connection之上的⼀个抽象层. 在 RabbitMQ 中, ⼀个TCP连接可以有多个Channel, 每个Channel 都是独⽴的虚拟连接. 消息的发送和接收都是基于 Channel的.
通道的主要作⽤是将消息的读写操作复⽤到同⼀个TCP连接上，这样可以减少建⽴和关闭连接的开销, 提⾼性能.

2.3 Virtual host

Virtual host: 虚拟主机. 这是⼀个虚拟概念. 它为消息队列提供了⼀种逻辑上的隔离机制. 对于RabbitMQ而言, ⼀个 BrokerServer 上可以存在多个 Virtual Host. 当多个不同的⽤⼾使⽤同⼀个RabbitMQ Server 提供的服务时，可以虚拟划分出多个 vhost，每个用户在自己的 vhost 创建 exchange/queue 等

类似MySQL的\"database\", 是⼀个逻辑上的集合. ⼀个MySQL服务器可以有多个database.

2.4 Queue

Queue: 队列, 是RabbitMQ的内部对象, 用于存储消息.

【RabbitMQ】 RabbitMQ快速上手_rabbitmq详细教程
多个消费者, 可以订阅同⼀个队列

2.5 Exchange

Exchange: 交换机. message 到达 broker 的第⼀站, 它负责接收生产者发送的消息, 并根据特定的规则把这些消息路由到⼀个或多个Queue列中.
Exchange起到了消息路由的作用，它根据类型和规则来确定如何转发接收到的消息.

类似于发快递之后, 物流公司怎么处理呢, 根据咱们的地址来分派这个快递到不同的站点, 然后再送到收件⼈手里. 这个分配的⼯作,就是交换机来做的

2.6 RabbitMQ工作流程

理解了上⾯的概念之后, 再来回顾⼀下这个图, 来看RabbitMQ的工作流程：

【RabbitMQ】 RabbitMQ快速上手_rabbitmq详细教程

Producer 生产了⼀条消息
Producer 连接到RabbitMQBroker, 建⽴⼀个连接(Connection),开启⼀个信道(Channel)
Producer 声明⼀个交换机(Exchange), 路由消息
Producer 声明⼀个队列(Queue), 存放信息
Producer 发送消息⾄RabbitMQ Broker
RabbitMQ Broker 接收消息, 并存入相应的队列(Queue)中, 如果未找到相应的队列, 则根据⽣产者的配置, 选择丢弃或者退回给生产者。

如果我们把RabbitMQ比作⼀个物流公司，那么它的⼀些核心概念可以这样理解：
1 . Broker就类似整个物流公司的总部, 它负责协调和管理所有的物流站点, 确保包裹安全、⾼效地送达.
2 . Virtual Host可以看作是物流公司为不同的客⼾或业务部⻔划分的独⽴运营中⼼. 每个运营中⼼都有⾃⼰的仓库(Queue), 分拣规则(Exchange)和运输路线(Connection和Channel), 这样可以确保不同客⼾的包裹处理不会相互干扰, 同时提供定制化的服务
3 . Exchange就像是站点里的分拣中心. 当包裹到达时, 分拣中心会根据包裹上的标签来决定这个包裹应该送往哪个目的地(队列). 快递站点可能有不同类型的分拣中⼼, 有的按照具体地址分拣, 有的将包裹复制给多个收件⼈等.
4 . Queue就是快递站点里的⼀个个仓库, ⽤来临时存放等待派送的包裹. 每个仓库都有⼀个或多个快递员(消费者)负责从仓库中取出包裹并派送给最终的收件人.
5 . Connection就像是快递员与快递站点之间的通信线路. 快递员需要通过这个线路来接收派送任务 (消息).
6 . Channel就像是快递员在执行任务时使用的多个并行的通信线路. 这样，快递员可以同时处理多个包裹, 比如⼀边派送包裹, ⼀边接收新的包裹

二、AMQP协议

AMQP（Advanced Message Queuing Protocol）是⼀种高级消息队列协议, AMQP定义了⼀套确定的消息交换功能, 包括交换器(Exchange), 队列(Queue) 等. 这些组件共同工作, 使得生产者能够将消息发送到交换器. 然后由队列接收并等待消费者接收. AMQP还定义了⼀个网络协议, 允许客户端应用通过该协议与消息代理和AMQP模型进行交互通信
RabbitMQ是遵从AMQP协议的,换句话说,RabbitMQ就是AMQP协议的Erlang的实现. AMQP的模型结构和RabbitMQ的模型结构是⼀样的.

三、web界面操作

RabbitMQ管理界面上的Connections,Channels, Exchange, Queues 就是和上⾯流程图的概念是⼀样的, Overview就是视图的意思, Admin是用户管理. 我们在操作RabbitMQ前, 需要先创建Virtual host
接下来看具体操作：

4.1 用户相关操作

添加用户

a) 点击 Admin -> Add user

【RabbitMQ】 RabbitMQ快速上手_rabbitmq详细教程

b) 设置账号密码及权限
【RabbitMQ】 RabbitMQ快速上手_rabbitmq详细教程

①: 设置账号
②: 设置密码
③: 确认密码
④: 设置权限
添加完成后, 点击[Add user]

c) 观察用户是否添加成功
【RabbitMQ】 RabbitMQ快速上手_rabbitmq详细教程
用户相关操作：
a) 点击要删除的用户, 查看用户详情：

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

设置对虚拟机的操作权限
【RabbitMQ】 RabbitMQ快速上手_rabbitmq详细教程
更新/删除用户：

退出当前用户：

【RabbitMQ】 RabbitMQ快速上手_rabbitmq详细教程

4.2 虚拟主机相关操作

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

此操作会为当前登录用户设置虚拟主机

四、RabbitMQ快速入门

步骤:

引入依赖
编写生产者代码
编写消费者代码

4.1 引入依赖

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

4.2 编写生产者代码

4.2.1 创建连接

注意：RabbitMQ 默认的用于客户端连接的TCP 端口号是5672, 需要提前进行开放

// 1. 创建连接⼯⼚ConnectionFactory factory = new ConnectionFactory();//2. 设置参数factory.setHost(\"110.41.51.65\");//ip 默认值localhostfactory.setPort(5672); //默认值5672 factory.setVirtualHost(\"bite\");//虚拟机名称, 默认 /factory.setUsername(\"study\");//⽤⼾名,默认guestfactory.setPassword(\"study\");//密码, 默认guest//3. 创建连接ConnectionConnection connection = factory.newConnection();

4.2.2 创建Channel

注意：生产者和消费者创建的channel并不是同⼀个

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

4.2.3 声明⼀个队列Queue

/*queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments) 1.queue: 队列名称 2.durable: 是否持久化.true-设置队列为持久化, 待久化的队列会存盘,服务器重启之后, 消息不 丢失。 3.exclusive: * 是否独占, 只能有⼀个消费者监听队列 * 当Connection关闭时, 是否删除队列 4.autoDelete: 是否⾃动删除, 当没有Consumer时, ⾃动删除掉 5.arguments: ⼀些参数 *///如果没有⼀个hello 这样的⼀个队列, 会⾃动创建, 如果有, 则不创建channel.queueDeclare(\"hello\",true,false,false,null);

4.2.4 发送消息

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

【RabbitMQ】 RabbitMQ快速上手_rabbitmq详细教程

例如: 如果有⼀个名为 “hello” 的队列, 生产者可以直接发送消息到 “hello” 队列, 而消费者可以从\"hello\" 队列中接收消息, ⽽不需要关心交换机的存在. 这种模式⾮常适合简单的应⽤场景，其中生产者和消费者之间的通信是⼀对⼀的.

//6. 通过channel发送消息到队列中/*basicPublish(String exchange, String routingKey, AMQP.BasicProperties props, byte[] body)1.exchange: 交换机名称, 简单模式下, 交换机会使⽤默认的\"\"2.routingKey: 路由名称, routingKey = 队列名称3.props: 配置信息4.body: 发送消息的数据 */String msg = \"Hello World\";//使⽤的是内置交换机. 使⽤内置交换机时, routingKey要和队列名称⼀样, 才可以路由到对应的队##列上去channel.basicPublish(\"\",\"hello\",null,msg.getBytes());System.out.println(msg + \"消息发送成功\");

4.2.5 释放资源

//显式地关闭Channel是个好习惯, 但这不是必须的, Connection关闭的时候,Channel也会⾃动关闭.channel.close();connection.close();

4.2.6 观察结果

运行之前：
【RabbitMQ】 RabbitMQ快速上手_rabbitmq详细教程

运行之后, 队列中就已经有了hello这个队列的信息：

【RabbitMQ】 RabbitMQ快速上手_rabbitmq详细教程

如果在代码中注掉资源释放的代码, 在Connections和Channels也可以看到相关信息

【RabbitMQ】 RabbitMQ快速上手_rabbitmq详细教程

Queue也可以配置显示Consumer相关信息

【RabbitMQ】 RabbitMQ快速上手_rabbitmq详细教程

4.3 编写消费者代码

消费者代码和⽣产者前3步都是⼀样的, 第4步改为消费当前队列

创建连接
创建Channel
声明⼀个队列Queue
消费消息
释放资源

4.3.1 消费当前队列

basicConsume

/*basicConsume(String queue, boolean autoAck, Consumer callback)参数:1. queue: 队列名称2. autoAck: 是否⾃动确认, 消费者收到消息之后,⾃动和MQ确认3. callback: 回调对象 */String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;

Consumer

Consumer ⽤于定义消息消费者的行为. 当我们需要从RabbitMQ接收消息时, 需要提供⼀个实现了Consumer 接口的对象.
DefaultConsumer 是 RabbitMQ提供的⼀个默认消费者, 实现了Consumer 接口.

核心方法:

handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) : 从队列接收到消息时, 会自动调用该方法. 在这个方法中, 我们可以定义如何处理接收到的消息, 例如打印消息内容, 处理业务逻辑或者将消息存储到数据库等. 参数说明如下：
▪ consumerTag : 消费者标签, 通常是消费者在订阅队列时指定的.
▪ envelope : 包含消息的封包信息，如队列名称, 交换机等.
▪ properties : ⼀些配置信息
▪ body : 消息的具体内容

//6. 接收消息, 并消费/*basicConsume(String queue, boolean autoAck, Consumer callback)参数:1. queue: 队列名称2. autoAck: 是否⾃动确认, 消费者收到消息之后,⾃动和MQ确认3. callback: 回调对象 */DefaultConsumer consumer = new DefaultConsumer(channel) { /* 回调⽅法, 当收到消息后, 会⾃动执⾏该⽅法 1. consumerTag: 标识 2. envelope: 获取⼀些信息, 交换机, 路由key 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);

4.3.2 释放资源

//等待回调函数执⾏完毕之后, 关闭资源TimeUnit.SECONDS.sleep(5);//7. 释放资源 消费者相当于是⼀个监听程序, 不需要关闭资源channel.close();connection.close();

实际上消费者相当于是⼀个监听程序, 不需要关闭资源

4.3.3 观察结果

运⾏程序, 我们刚才发送的消息, 就收到了

接收到消息: Hello World

如果我们不释放资源, 可以看到响应的Connection, channel

【RabbitMQ】 RabbitMQ快速上手_rabbitmq详细教程

4.4 源码

生产者代码：

import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;public class RabbitProducer { public static void main(String[] args) throws Exception { // 1. 创建连接⼯⼚ ConnectionFactory factory = new ConnectionFactory(); //2. 设置参数 factory.setHost(\"110.41.51.65\");//ip 默认值localhost factory.setPort(15673); //默认值5672 factory.setVirtualHost(\"bite\");//虚拟机名称, 默认 / factory.setUsername(\"study\");//⽤⼾名,默认guest factory.setPassword(\"study\");//密码, 默认guest //3. 创建连接Connection Connection connection = factory.newConnection(); //4. 创建channel通道 Channel channel = connection.createChannel(); //5. 声明队列 /* queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments) 1.queue: 队列名称 2.durable: 是否持久化, 当mq重启之后, 消息还在 3.exclusive: * 是否独占, 只能有⼀个消费者监听队列 * 当Connection关闭时, 是否删除队列 4.autoDelete: 是否⾃动删除, 当没有Consumer时, ⾃动删除掉 5.arguments: ⼀些参数 */ //如果没有⼀个hello 这样的⼀个队列, 会⾃动创建, 如果有, 则不创建 channel.queueDeclare(\"hello\", true, false, false, null); //6. 通过channel发送消息到队列中 /* basicPublish(String exchange, String routingKey, AMQP.BasicProperties props, byte[] body) 1. exchange: 交换机名称, 简单模式下, 交换机会使⽤默认的\"\" 2.routingKey: 路由名称, routingKey = 队列名称 3.props: 配置信息 4.body: 发送消息的数据 */ String msg = \"Hello World\"; //使⽤的是内置交换机. 使⽤内置交换机时, routingKey要和队列名称⼀样, 才可以路由 到对应的队列上去 channel.basicPublish(\"\", \"hello\", null, msg.getBytes()); //7.释放资源 System.out.println(msg + \"消息发送成功\"); channel.close(); connection.close(); }}

消费者代码：

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\");//ip 默认值localhost factory.setPort(15673); //默认值5672 factory.setVirtualHost(\"bite\");//虚拟机名称, 默认 / factory.setUsername(\"study\");//⽤⼾名,默认guest factory.setPassword(\"study\");//密码, 默认guest //3. 创建连接Connection Connection connection = factory.newConnection(); //4. 创建channel通道 Channel channel = connection.createChannel(); //5. 声明队列 /* queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments) 1.queue: 队列名称 2.durable: 是否持久化, 当mq重启之后, 消息还在 3.exclusive: * 是否独占, 只能有⼀个消费者监听队列 * 当Connection关闭时, 是否删除队列 4.autoDelete: 是否⾃动删除, 当没有Consumer时, ⾃动删除掉 5.arguments: ⼀些参数 */ //如果没有⼀个hello 这样的⼀个队列, 会⾃动创建, 如果有, 则不创建 channel.queueDeclare(\"hello\", true, false, false, null); //6. 接收消息, 并消费 /* basicConsume(String queue, boolean autoAck, Consumer callback) 参数: 1. queue: 队列名称 2. autoAck: 是否⾃动确认, 消费者收到消息之后,⾃动和MQ确认 3. callback: 回调对象 */ DefaultConsumer consumer = new DefaultConsumer(channel) { /* 回调⽅法, 当收到消息后, 会⾃动执⾏该⽅法 1. consumerTag: 标识 2. envelope: 获取⼀些信息, 交换机, 路由key 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(); }}

结语

本文主要介绍了RabbitMQ的核心概念并简单介绍了RabbitMQ的管理页面，之后介绍了RabbitMQ的相关界面操作。最后编写了一个简单的生产消费代码。

以上就是本文全部内容，感谢各位能够看到最后，如有问题，欢迎各位大佬在评论区指正，希望大家可以有所收获！创作不易，希望大家多多支持！

最后，大家再见！祝好！我们下期见！

【RabbitMQ】 RabbitMQ快速上手_rabbitmq详细教程

文章目录

一、RabbitMQ 核心概念

1.1 Producer和Consumer

2.2 Connection和Channel

2.3 Virtual host

2.4 Queue

2.5 Exchange

2.6 RabbitMQ工作流程

二、AMQP协议

三、web界面操作

4.1 用户相关操作

4.2 虚拟主机相关操作

四、RabbitMQ快速入门

4.1 引入依赖

4.2 编写生产者代码

4.2.1 创建连接

4.2.2 创建Channel

4.2.3 声明⼀个队列Queue

4.2.4 发送消息

4.2.5 释放资源

4.2.6 观察结果

4.3 编写消费者代码

4.3.1 消费当前队列

4.3.2 释放资源

4.3.3 观察结果

4.4 源码

结语

公告

DeepSeek全套部署资料免费下载

免费可商用字体批量下载

【RabbitMQ】 RabbitMQ快速上手_rabbitmq详细教程

文章目录

一、RabbitMQ 核心概念

1.1 Producer和Consumer

2.2 Connection和Channel

2.3 Virtual host

2.4 Queue

2.5 Exchange

2.6 RabbitMQ工作流程

二、AMQP协议

三 、web界面操作

4.1 用户相关操作

4.2 虚拟主机相关操作

四、RabbitMQ快速入门

4.1 引入依赖

4.2 编写生产者代码

4.2.1 创建连接

4.2.2 创建Channel

4.2.3 声明⼀个队列Queue

4.2.4 发送消息

4.2.5 释放资源

4.2.6 观察结果

4.3 编写消费者代码

4.3.1 消费当前队列

4.3.2 释放资源

4.3.3 观察结果

4.4 源码

结语

相关问题

公告

DeepSeek全套部署资料免费下载

免费可商用字体批量下载

三、web界面操作