RabbitMQ安装
参考：RabbitMQ 安装，配置，java接入使用（详细教程）_rabbitmq安装-CSDN博客
RabbitMQ核心概念
在安装完RabbitMQ之后，我们接下来学习如何去使用RabbitMQ。
在上个篇幅中我们讲了RabbitMQ的安装，并安装了管理界面。
操作rabbitmq可以有三种方式：
1、通过代码来操作 5672
2、通过界面操作 15672
3、通过命令操作 rabbitmqctl
管理页面访问：http://主机IP:15672
RabbitMQ工作流程图：
ps：
Exchange：交换机
Queue：队列
生产者和消费者都是RabbitMQ服务器的客户端。一个Connection可以有多个Channel；一个Broker可以有多个虚拟（主）机，这些虚拟机逻辑上是隔离的。
RabbitMQ是一个消息中间件，也是一个生产者消费者模型，负责接收、存储、转发消息。
消息传递的过程类似于邮局：
当你要发送一个邮件时，你把你的邮件发到邮局，邮件收到邮件，并通过邮递员送到收件人手上。
按照这个逻辑，生产者就相当于发件人，邮局就相当于RabbitMQ服务器，收件人就相当于消费者。
Producer和Consumer
Producer：生产者，是RabbitMQ Server的客户端，向RabbitMQ发送消息。
Consumer：消费者，也是RabbitMQ Server的客户端，从RabbitMQ接收消息。
Broker：其实就是RabbitMQ Server，主要是接收和转发消息。
生产者创建消息，然后发布到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连接上，这样可以减少建立和关闭连接的开销，提高性能。
Virtual host
Virtual host：虚拟主机，这是一个虚拟概念。它为消息队列提供了一种逻辑上的隔离机制。对于RabbitMQ而言，一个BrokerSever上可以存在多个Virtual Host。当多个不同的用户使用同一个RabbitMQ Server提供的服务时，可以虚拟划分出多个vhost，每个用户在自己的vhost创建exchange/queue等。
类似于MySQL的database，是逻辑上的集合。一个MySQL服务器上可以有多个database
Queue
Queue：队列，是RabbitMQ的内部对象，用于存储消息。
多个消费者，可以订阅同一个队列。他们的关系是多对多的，一个队列可以有多个消费者订阅的；一个消费者也可以订阅多个队列：
Exchange
Exchange：交换机。message到达broker的第一站，它负责接收生产者发送的消息，并根据特定的规则，把这些消息路由到一个或者多个queue中。
Exchange起到了消息路由的作用，它根据规则来确定如何转发接收到的消息。
类似于发快递之后，快递公司怎么处理呢，根据咱们的地址来分派这个快递到达不同的站点，然后再送到收件人手中。这个分配的工作，就是交换机来做的。
RabbitMQ工作流程
如果我们把RabbitMQ比作一个物流公司，那么它的核心概念：
1、Broker就类似于整个物流公司的总部，它负责协调和管理所有的物流站点，确保包裹安全、高效地送达。
2、Virtual Host可以看作是物流公司为不同客户或业务部门划分的独立运营中心，每个运营中心都有自己的仓库（Queue），分拣规则（Exchange）和运输路线（Connectional和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的实现（当然RabbitMQ还支持STOMP2,MQTT2等协议）。AMQP模型结构和RabbitMQ的模型结构是一样的。
管理界面操作
RabbitMQ管理界面上的Connections，Channels，Exchange，Queues就是和上面流程图的概念是一样的，Overview就是视图的意思，Admin是用户管理。
我们在操作RabbitMQ前，需要先创建Virtual host。
接下来我们开始具体操作：
用户相关操作
添加用户
1、点击Admin->Add user
2、观察用户是否添加成功
为用户设置权限
点击进入用户管理界面：
选择要使用的虚拟机并进行设置（这里的/表示所有的虚拟机，即任何虚拟机，xmy都有权限操作）
更新/删除用户
退出当前用户
虚拟主机相关操作
设置虚拟主机名称
观察设置结果：
RabbitMQ快速入门
步骤：
1、引入依赖
2、编写生产者代码
3、编写消费者代码
引入依赖
<dependencies>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.7.3</version>
</dependency>
</dependencies>
编写生产者代码
建立连接
//1、建立连接
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost(你的主机IP);
connectionFactory.setPort(5672);//需要提前开放端口号
connectionFactory.setUsername("admin");//账号
connectionFactory.setPassword("admin");//密码
connectionFactory.setVirtualHost("xmy");//虚拟主机
Connection connection = connectionFactory.newConnection();
开启信道
//2、开启信道
Channel channel = connection.createChannel();
声明交换机
这一步我们使用系统内置的交换机即可，无需声明。
声明队列
参数介绍：（按照先后排序）
queue：队列名称
durable：可持久化
exclusive：是否独占
autoDelete：是否自动删除
argument：参数
//4、声明队列
/**
* queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) throws IOException;
* 参数介绍：
* queue：队列名称
*durable：可持久化
* exclusive：是否独占
* autoDelete：是否自动删除
* argument：参数
*/
channel.queueDeclare("hello",true,false,false,null);
发送消息
参数介绍：
exchange：交换机名称
rountingKey：内置交换机，rountingKey和队列名称保持一致
props：属性配置
body：消息体
//5、发送消息
/**
* basicPublish（String exchange，String routingKey,BasicProperties prop,byte[] body)
* 参数说明：
* exchange：交换机名称
* routingKey：内置交换机，routingKey和队列名称保持一致
* props:属性配置
* body：消息体
*/
String msg = "hello,xmy!";
channel.basicPublish("","hello",null,msg.getBytes());
System.out.println("消息发送成功！！！");
资源释放
注意：这里是先释放信道，再释放连接！
//6、资源释放
channel.close();
connection.close();
运行程序：
可以看到在xmy这个已经存在一条消息，此时由于还没有消费者，所以还没被消费。
如果在代码中注掉资源释放的代码，在Connection和Channels也可以看到相关信息：
Queue也可以配置显示Consumer相关信息：
编写消费者代码
创建连接
//1、创建连接
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost(你的主机ip);
connectionFactory.setPort(5672);//需要提前开放端口号
connectionFactory.setUsername("admin");//账号
connectionFactory.setPassword("admin");//密码
connectionFactory.setVirtualHost("xmy");//虚拟主机
Connection connection = connectionFactory.newConnection();
创建channel
//2、创建channel
Channel channel = connection.createChannel();
声明队列
//3、声明队列（可以省略）
channel.queueDeclare("hello",true,false,false,null);
消费消息
参数说明：
queue：队列名称
autoAck：是否自动确认
callback：接收消息后执行的逻辑是啥
//4、消费消息
/**
* basicConsume(String queue，boolean autoAck，Consumer callback)
* 参数说明：
* queue：队列名称
* autoAck：是否自动确认
* callback：接收消息后执行的逻辑是啥
*/
DefaultConsumer consumer = new DefaultConsumer(channel){
//从队列种收到消息的方法
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//TODO
System.out.println("接收到消息："+ new String(body));
}
};
channel.basicConsume("hello",true,consumer);
ps：这里我们可以让进程等待两秒，方便观察打印结果：
Thread.sleep(2000);
释放资源
//5、释放资源
channel.close();
connection.close();
运行程序：
此时，可以看到我们刚才发送的消息被收到了：
队列中存在的消息也消失了：
可能遇到的一些报错
1、资源释放顺序反了
此时我们先释放连接再释放信道：
connection.close();
channel.close();
报错信息：
2、队列不存在
我们把消费者声明队列这一步注掉：
然后再把hello这个队列删除掉：
报错信息：
3、IP/端口错误
此处我们将ip/端口改错，报错信息：
4、账号/密码错误
将账号/密码改错：
报错信息：
5、用户对该虚拟机没有操作权限
让当前的admin用户对虚拟机没有权限：
运行程序，报错：
生产者完整代码
public class ProducerDemo {
public static void main(String[] args) throws IOException, TimeoutException {
//1、建立连接
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost(云服务器ip);
connectionFactory.setPort(5672);//需要提前开放端口号
connectionFactory.setUsername("admin");//账号
connectionFactory.setPassword("admin");//密码
connectionFactory.setVirtualHost("xmy");//虚拟主机
Connection connection = connectionFactory.newConnection();
//2、开启信道
Channel channel = connection.createChannel();
//3、声明交换机 使用内置的交换机
//4、声明队列
/**
* queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) throws IOException;
* 参数介绍：
* queue：队列名称
*durable：可持久化
* exclusive：是否独占
* autoDelete：是否自动删除
* argument：参数
*/
channel.queueDeclare("hello",true,false,false,null);
//5、发送消息
/**
* basicPublish（String exchange，String routingKey,BasicProperties prop,byte[] body)
* 参数说明：
* exchange：交换机名称
* routingKey：内置交换机，routingKey和队列名称保持一致
* props:属性配置
* body：消息体
*/
String msg = "hello,xmy!";
channel.basicPublish("","hello",null,msg.getBytes());
System.out.println("消息发送成功！！！");
//6、资源释放
channel.close();
connection.close();
}
}
消费者完整代码
public class ComsumerDemo {
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
//1、创建连接
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost(云服务器ip);
connectionFactory.setPort(5672);//需要提前开放端口号
connectionFactory.setUsername("admin");//账号
connectionFactory.setPassword("admin");//密码
connectionFactory.setVirtualHost("xmy");//虚拟主机
Connection connection = connectionFactory.newConnection();
//2、创建channel
Channel channel = connection.createChannel();
//3、声明队列（可以省略）
channel.queueDeclare("hello",true,false,false,null);
//4、消费消息
/**
* basicConsume(String queue，boolean autoAck，Consumer callback)
* 参数说明：
* queue：队列名称
* autoAck：是否自动确认
* callback：接收消息后执行的逻辑是啥
*/
DefaultConsumer consumer = new DefaultConsumer(channel){
//从队列种收到消息的方法
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//TODO
System.out.println("接收到消息："+ new String(body));
}
};
channel.basicConsume("hello",true,consumer);
//等待程序执行完成
Thread.sleep(2000);
//5、释放资源
channel.close();
connection.close();
}
}
