宜昌百度网站建设龙口网站设计
目录
MQ简介
1、同步通信
图片
2、异步通信
图片
RabbitMQ快速上手
基本介绍：
Producer和Consumer
Connection和Channel
Virtual host
Queue
Exchange
工作流程
AMQP
Java编写RabbitMQ生产者消费者
生产者
1.建立连接
2.开启信道
3.声明交换机
4.声明队列
5.发送消息
6.资源释放
生产者全部代码：
消费者
执行结果
MQ简介
MQ(Message Queue)消息队列，本质上是队列，满足队列FIFO(先入先出)的性质,队列中存放的内容是消息，消息可以是只包含文本字符串，JSON，也可以是对象等。
MQ经常用于分布式系统之间的通信，系统之间的通信通常有两种方式：
1、同步通信
直接调用对方的服务，数据从一端出发到达另一端。
图片
2、异步通信
数据从⼀端发出后，先进⼊⼀个容器进⾏临时存储，当达到某种条件后，再由这个容器发送给另⼀端，这个容器通常就是MQ(message queue)。
图片
RabbitMQ快速上手
基本介绍：
RabbitMQ是MQ的一种实现，工作流程如下：
RabbitMQ是一个消息中间件，也是一个生产者消费者模型，负责接收，存储并转发消息，消息传递的过程类似于发快递，你把快递放到驿站，快递小哥帮你把快递送到接收人的手上。这个过程中，你就相当于Producer，接收人相当于Consumer，快递站就是RabbitMQ。
Producer和Consumer
Producer(生产者)，是RabbitMQ的客户端，用于发送消息到RabbitMQ。
Consumer(消费者)，也是RabbitMQ的客户端，用于向RabbitMQ接收消息。
Broker(代理)，RabbitMQ服务器节点，用于接收和发送消息。
生产者发送消息到RabbitMQ服务器，让RabbitMQ进行路由转发到对应的消费者。
生产者发送的消息带有标签，代理（RabbitMQ服务器）会根据标签路由，找到需要的消费者，进行消息转发。消费者接收消息进行消费过程中标签就会被丢掉，也就是说消费者并不会知道消息的发送者是谁。
Connection和Channel
Connection(连接)：客户端和RabbitMQ服务器之间的⼀个TCP连接，建立连接后才可以进行消息的发送和接收。
Channel(信道)：类似于发送消息的一个通道，每个TCP连接可以有多个信道，每个信道都是独立的虚拟连接，消息的发送和接收都是基于Channel(信道)的。
Virtual host
Virtual host：虚拟主机，给消息队列提供逻辑上的隔离，一个BrokerServer可以有多个虚拟主机，当不同用户使用RabbitMQ Server提供的服务时，可以使用划分多个虚拟主机的方式将一系列业务隔离开来，类似于MySQL中不同的数据库。
Queue
Queue：队列，用于存放消息，一个队列可以被多个消费者订阅。
Exchange
Exchenge：交换机，交换机负责接收生产者发送的消息，并按照规则将消息路由到队列上。类似于快递小哥将快递送给
工作流程
Proucer生产消息 -> Producer连接到RabbitMQBroker，建立连接(Connection)，开启信道(Channel) -> Producer声明交换机(Exchange),路由消息 ->Producer声明队列(Queue) -> Produce发送消息到RabbitMQBroker-> RabbitMQBroker接收消息，并放入相应的队列(Queue)中，未找到相应队列根据生产者的配置，选择丢弃或者退回给生产者。
AMQP
AMQP是一种高级消息队列协议，定义了一套确定的消息交换功能，包括交换机，队列等，这些组件共同工作，使生产者能够将消息发送到交换机，由队列接收并等待消费者接收。AMQP还定义了一个网络协议，允许客户端应用通过这个协议与消息代理和AMQP模型进行交互通信。RabbitMQ使AMQP协议的Erlang实现。
Java编写RabbitMQ生产者消费者
打开管理页面，添加用户：
这里添加的测试用户名称为：test 密码 123456。
创建虚拟机：
这里添加的虚拟机名称为：test
返回admin页面，点击刚刚创建的test用户，添加虚拟机操作权限。
进入IDEA创建一个Maven项目，导入依赖：
<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.20.0</version>
</dependency>生产者
1.建立连接
使用依赖中的ConnectionFactory类创建连接，设置关键信息：
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("139.9.116.124");
connectionFactory.setPort(5672);
connectionFactory.setUsername("fwx");
connectionFactory.setPassword("123456");
connectionFactory.setVirtualHost("测试");
Connection connection = connectionFactory.newConnection();2.开启信道
Channel channel = connection.createChannel();3.声明交换机
这里使用的是RabbitMQ默认提供的交换机，不需要代码进行声明
4.声明队列
使用channel的queueDeclare方法，依此填入相关参数：
channel.queueDeclare("hello",true,false,false,null);打开源码，查看参数的含义：
第一个参数queue代表队列的名称，第二个参数durable代表是否持久化（true-设置队列为持久化, 待久化的队列会存盘,服务器重启之后, 消息不丢失），第三个参数exclusive代表是否独占（是否独占, 只能有⼀个消费者监听队列），第四个参数autoDelete代表是否自动删除（没有消费者自动删除队列），第五个参数arguments代表一些配置的参数。
5.发送消息
使用Channel的basicPublish()方法进行消息的发送，这里发送的消息为"hello RabbitMQ~"，消息发送10次：
for (int i = 0; i < 10; i++) {String msg = "hello RabbitMQ~";channel.basicPublish("","test1",null,msg.getBytes());
}打开方法，分析方法参数：
第一个参数exchange代表交换机的名称(这里使用默认交换机"")，第二个参数routingKey代表路由的名称（使用默认交换机，路由名称要和队列名称相同才可以找到对应队列），第三个参数body代表发送的消息体（将String类型转换为字节数组）。
6.资源释放
channel.close();
connection.close();最后将资源关闭即可。
生产者全部代码：
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 Producer {public static void main(String[] args) throws IOException, TimeoutException {//1.建立链接ConnectionFactory connectionFactory = new ConnectionFactory();//主机地址connectionFactory.setHost("139.9.116.124");//端口号connectionFactory.setPort(5672);//用户名connectionFactory.setUsername("test");//密码connectionFactory.setPassword("123456");//虚拟机connectionFactory.setVirtualHost("test");//连接建立Connection connection = connectionFactory.newConnection();//2.开启信道Channel channel = connection.createChannel();//3.声明交换机,使用内置交换机//4.声明队列channel.queueDeclare("test1",true,false,false,null);//5.发送消息for (int i = 0; i < 10; i++) {String msg = "hello RabbitMQ~";channel.basicPublish("","test1",null,msg.getBytes());}//6.资源释放channel.close();connection.close();}
}消费者
消费者与生产者逻辑相似，这里不再讲解，全部代码：
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Consumer {public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {//1.建立链接ConnectionFactory connectionFactory = new ConnectionFactory();//主机地址connectionFactory.setHost("139.9.116.124");//端口号connectionFactory.setPort(5672);//用户名connectionFactory.setUsername("test");//密码connectionFactory.setPassword("123456");//虚拟机connectionFactory.setVirtualHost("test");//连接建立Connection connection = connectionFactory.newConnection();//2.开启信道Channel channel = connection.createChannel();//3.声明队列(生产者已经声明)//4.消费消息/**参数声明* basicConsume(String queue, boolean autoAck, Consumer callback)* queue:队列名* autoAck:自动确认* callback：接收到消息执行逻辑*/DefaultConsumer consumer = new DefaultConsumer(channel){//从队列中收到消息，就会执行的方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//TODOSystem.out.println("接收的消息："+new String(body));}};channel.basicConsume("test1",true,consumer);//等待程序执行完成Thread.sleep(2000);//5.释放资源channel.close();connection.close();}
}
生产者已经声明过队列，消费者无需声明，消费者通过重写handleDelivery()方法（这个方法在接收消息时自动执行）去消费消息。
执行结果
执行生产者代码，打开RabbitMQ管理页面，看到队列test1产生了10条消息等待消费：
执行消费者，此时控制台打印了接收的消息：
再查看管理页面，消息已经全部消费：
到此结束，感谢观看！