当前位置：首页 > news >正文

3、初识RabbitMQ

news 来源：原创 2025/4/28 10:44:48

界面上的导航栏共分6部分，分别代表不同的意思

一、Producer和Consumer

Producer: 生产者, 是RabbitMQ Server的客户端, 向RabbitMQ发送消息

Consumer: 消费者, 也是RabbitMQ Server的客⼾端, 从RabbitMQ接收消息

Broker：其实就是RabbitMQ Server, 主要是接收和收发消息

生产者（Producer）创建消息，然后发布到RabbitMQ中，在实际应用中，消息通常是一个带有一定业务逻辑结构的数据，比如JSON字符串，消息可以带有一定的标签，RabbitMQ会根据标签进行路由，把消息发送给感兴趣的消费者（Cosumer）

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

对于RabbitMQ来说，一个 RabbitMQ Breaker 可以简单地看做一个RabbitMQ服务器节点，或者RabbitMQ服务实例，大多数情况下也可以将一个RabbitMQ Breaker看做一台RabbitMQ服务器

二、Connetion和Channel

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

Channel: 通道，信道 Channel 是在Connection之上的一个抽象层，在RabbitMQ，一个TCP连接可以有多个Channel，每个Channel 都是独立的虚拟连接消息的发送和接收都是基于Channel的。通到的主要作用是将消息的读写操作复用到同一个TCP连接上，这样可以减少建立和关闭连接的开销，提升性能。

三、Virtual host

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

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

四、Queue

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

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

五、Exchange

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

RabbitMQ是遵从AMQP协议的,换句话说,RabbitMQ就是AMQP协议的Erlang的实现(当然RabbitMQ还⽀持STOMP2, MQTT2等协议). AMQP的模型结构和RabbitMQ的模型结构是⼀样的

五、初识RabbitMQ

步骤：

a 引入pom依赖

b 生产者代码

c 消费者代码

5.1 引入pom依赖

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

5.2 生产者代码

package rabbitmq;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class ProducerDemo {public static void main(String[] args) throws IOException, TimeoutException {//1、建立连接// 创建一个ConnectionFactory实例来配置RabbitMQ连接ConnectionFactory connectionFactory = new ConnectionFactory();// 设置RabbitMQ服务器的主机地址connectionFactory.setHost("8.136.108.248");// 设置RabbitMQ服务器的端口号connectionFactory.setPort(5672);// 设置登录RabbitMQ服务器的用户名connectionFactory.setUsername("pinkboy");// 设置登录RabbitMQ服务器的密码connectionFactory.setPassword("123456");// 设置RabbitMQ服务器的虚拟主机connectionFactory.setVirtualHost("/");// 使用ConnectionFactory创建一个新的连接Connection connection = connectionFactory.newConnection();//2、创建通道Channel channel = connection.createChannel();//3、声明交换机//4、声明队列/*** 声明一个队列* * @param channel RabbitMQ的通道，用于执行队列操作* * 此处使用了queueDeclare方法来声明一个名为"hello"的队列该方法的参数分别表示：* 1. 队列名称("hello")：指定要声明的队列的名称* 2. true：表示该队列是持久化的，意味着即使RabbitMQ服务重启，队列也会被保留* 3. false：表示该队列不是排他的，意味着该队列可以被所有通道共享* 4. false：表示该队列不会在使用后自动删除，需要手动删除* 5. null：表示不设置额外的参数* * 选择这些参数值的原因可能是希望创建一个持久化的、共享的队列，以便在不同的时间点和不同的消费者之间传递消息*/channel.queueDeclare("hello", true, false, false, null);//5、发送消息// 循环发送消息到 RabbitMQ 的 "hello" 队列中for (int i = 0; i < 10; i++) {// 构造消息内容String msg = "hello rabbitmq~" + i;// 发布消息到队列，使用空交换机名称表示使用默认交换机// 第三个参数为消息的属性，null 表示使用默认属性// 将消息内容转换为字节数组作为消息体发送channel.basicPublish("", "hello", null, msg.getBytes());}System.out.println("消息发送成功!");//6、释放资源channel.close();connection.close();}
}

5.3 消费者代码

package rabbitmq;import com.rabbitmq.client.*;
import java.io.IOException;public class ConsumerDemo {public static void main(String[] args) throws Exception {//1.创建连接// 创建一个ConnectionFactory实例来配置RabbitMQ连接ConnectionFactory connectionFactory = new ConnectionFactory();// 设置RabbitMQ服务器的主机地址connectionFactory.setHost("8.136.108.248");// 设置RabbitMQ服务器的端口号connectionFactory.setPort(5672);// 设置登录RabbitMQ服务器的用户名connectionFactory.setUsername("pinkboy");// 设置登录RabbitMQ服务器的密码connectionFactory.setPassword("123456");// 设置RabbitMQ服务器的虚拟主机connectionFactory.setVirtualHost("/");// 使用ConnectionFactory创建一个新的连接Connection connection = connectionFactory.newConnection();//2.创建ChannelChannel channel = connection.createChannel();/***3.声明一个队列* * @param channel RabbitMQ的通道，用于执行队列操作* * 此处使用了queueDeclare方法来声明一个名为"hello"的队列该方法的参数分别表示：* 1. 队列名称("hello")：指定要声明的队列的名称* 2. true：表示该队列是持久化的，意味着即使RabbitMQ服务重启，队列也会被保留* 3. false：表示该队列不是排他的，意味着该队列可以被所有通道共享* 4. false：表示该队列不会在使用后自动删除，需要手动删除* 5. null：表示不设置额外的参数* * 选择这些参数值的原因可能是希望创建一个持久化的、共享的队列，以便在不同的时间点和不同的消费者之间传递消息*/channel.queueDeclare("hello", true, false, false, null);// 4.开始从名为"hello"的队列中消费消息channel.basicConsume("hello", true, new DefaultConsumer(channel) {/*** 处理接收到的消息* * @param consumerTag 消费者标签，用于标识消费者* @param envelope 包含消息路由信息的信封* @param properties 消息的属性，如内容类型、内容编码等* @param body 消息的主体内容，以字节数组形式表示* @throws IOException 如果处理消息时发生I/O错误*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {// 打印消息的主体内容System.out.println("body:" + new String(body));}});Thread.sleep(2000);//5.关闭资源channel.close();connection.close();}
}

观察运行结果：

运行生产者代码