eyb：RabbitMQ学习1

大家好，我是讯享网，很高兴认识大家。

目录：

（1）MQ介绍

（2）RabbitMQ介绍

（3）RabbitMQ的安装

（4）管理界面介绍

（5）简单队列

（6）工作队列-轮询

（7）工作队列-公平

（8）发布-订阅队列

消息的中间件：

讯享网

（1）MQ介绍

（2）RabbitMQ介绍

目前市面上主流的MQ：

ActiveMQ是Apache下的项目，是一个比较老的MQ了，前几年比较流行，现在不行了，因为它的性能没那么高，在小型的项目中使用，要求性能没那么高，但是在高并发的情况下，很容易出现消息的阻塞啊，错误啊，或者其他类型阻塞的情况，它的性能没那么好，我们发现现在啊大数据高并发这种情况到处可见，它就没办法在高并发大数据的情况下发挥好的性能所以现在不怎么用

Kafka：刚开始是用来做日志收集的，现在用来做消息的中间件，优点是性能比较高，缺点是对消息的可靠性没那么高，消息错误、重发、漏发等

RocketMQ：是阿里巴巴开元的消息中间件，在设计方面借鉴了Kafka，又做了改进，它具有Kafka的高性能，改进了Kafka消息错误啊可靠性的问题,虽然说开元了，但是有些功能还是不完整的

RabbitMQ:它的性能也比较不错，它也能保证消息的可靠性，它对高可用是非常厉害的，我们现在达到高可用都是搭建集群，主从，只要有任意一个节点有数据，它都能给你恢复过来，比如现在有三主三从，有一个节点磁盘里面有数据，能把整个集群的数据给你恢复过来，高可用非常高

（3）RabbitMQ的安装

以Linux环境安装

RabbitMQ是以Erlag语言编写的，要想安装需要先Erlang的环境，它两之间的版本要对应

下载安装包：

上传到Linux：

先安装Erlang：输入命令安转：

输入erl：出现这个就说明Erlang就安装成功了

在安装rabbitMQ：输入命令：

安装完毕之后启动：

查看启动状态：

向MySql安装完之后，它有Navcat可视化工具，RabbitMQ,也有可视化的：

可以准备一个RabbitMQ的可视化工具，是一个网页版，属于RabbitMQ的插件，插件怎么安装呢？

安装插件：

如果不能通过IP地址访问：可能是防火墙的问题:在linux暂时关闭防火墙。

然后就可以访问它可视化的页面

默认登录姓名密码：guest

访问出现错误，我们现在是跨域登录，默认只能本地登录

需要更改：配置

新建一个配置：

输入内容：

改完之后重启RubbitMQ：

重新登录：就可以登录成功了

（4）管理界面介绍

第一行是刷新

第二行是：虚拟路由，优点向Redis的虚拟的数据库

第三行是：集群默认的是单节点

第四行：用户

这里写图片描述

Connection：连接 Channels：信道 Exchange：交换机 Queues：队列 Consumer：消费者

Admin：用户

可以在添加有个用户：

点创建的用户，添加虚拟路由：

添加的用户没有虚拟路由

设置完之后，就有了

Virtual Host：虚拟路由

可以在添加一个虚拟路由：

再点击添加的路由，可以添加用户：

学习五种队列： 这里写图片描述

（5）简单队列

P：消息的生产者
C：消息的消费者
红色：队列

生产者将消息发送到队列，消费者从队列中获取消息。

发送消息：

接收消息：

生产者只要是用来发送消息，其他不进行处理，消费者去监听对应的队列，同时接收消息处理消息，消费之绑定的队列必须是同一个

创建Maven项目：选择模板：

加入依赖：

创建消息的生产者：

Send：

package com.xxxx.simple.send; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.nio.charset.StandardCharsets; /* * 简单队列：消息生产者 * * */ public class Send { //定义队列名称 private final static String QUEUE_NAME = "hello"; public static void main(String[] argv) throws Exception { // 创建连接工厂 ConnectionFactory factory=new ConnectionFactory(); //相关配置 factory.setHost("192.168.23.133"); factory.setUsername("yeb"); factory.setVirtualHost("/yeb"); factory.setPassword(""); factory.setPort(5672); try( //连接工厂创建连接 Connection connection=factory.newConnection(); //创建信道 Channel channel = connection.createChannel()){ // 绑定队列 参数第一个：队列名称 第二个：持久化 第三个:排他队列 第四个：自动删除 第五个：额外携带的参数 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 消息内容 String message = "Hello World!"; //发送消息 第一个参数是交换机，第三个：对应的参数 第四个：消息的实体 channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8)); System.out.println(" [x] Sent '" + message + "'"); } } }

讯享网

消息的消费之Recv：

讯享网package com.xxxx.simple.recv; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DeliverCallback; import java.nio.charset.StandardCharsets; /* * 简单队列：消息的消费者 * * */ public class Recv { //定义队列名称 private final static String QUEUE_NAME = "hello"; public static void main(String[] argv) throws Exception { // 创建连接工厂 ConnectionFactory factory=new ConnectionFactory(); //相关配置 factory.setHost("192.168.23.133"); factory.setUsername("yeb"); factory.setVirtualHost("/yeb"); factory.setPassword(""); factory.setPort(5672); //连接工厂创建连接 Connection connection=factory.newConnection(); //创建信道 Channel channel = connection.createChannel(); //绑定队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); System.out.println("[*] Waitint for message .To exit press CTRL+C"); DeliverCallback deliverCallback=(consumerTag,delivery)->{ String message=new String(delivery.getBody(),"UTF-8"); System.out.println(" [x] Received '" + message + "'"); }; //监听队列消费消息 channel.basicConsume(QUEUE_NAME,true,deliverCallback,consumerTag ->{ }); } }

启动消息的生产者：

发现队列里出现了hello消息：

启动消费者消费消息：

启动了消费者，消费者，一直处于连接的状态，它要一直去监听这个队列，并不会去关闭，而生产者不一样，是可以关闭的

信道是根基Connection创建的：

（6）工作队列-轮询

轮询：解决了消费能力跟不上生产能力的问题

现在有个问题，如果生产者每秒能生产100条消息，消费之能力有限，每秒只能收50条，每秒有50条消息在这里堆积，理论上消息队列是无限大的，但是它是基于你的内存和磁盘，当经过时间非常久之后，也会把你的磁盘撑破的RubbitMQ就没办法用了，这是它的一个缺点我们消费者的消费能力有高有低，不能匹配生产者的能力，这时候该怎么办呢？加消费者，这就讲到工作队列

工作队列分为两种模式：一种是轮询一种是公平

轮询：你一条我一条

公平：这个公平版不是所说的你一个我一个的公平

消息发送者：Send：

package com.xxxx.work.rr.send; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.nio.charset.StandardCharsets; /* * 工作队列（轮询）：消息的生产者 * * */ public class Send { //定义队列名称 private final static String QUEUE_NAME = "work_rr"; public static void main(String[] argv) throws Exception { // 创建连接工厂 ConnectionFactory factory=new ConnectionFactory(); //相关配置 factory.setHost("192.168.23.133"); factory.setUsername("yeb"); factory.setVirtualHost("/yeb"); factory.setPassword(""); factory.setPort(5672); try( //连接工厂创建连接 Connection connection=factory.newConnection(); //创建信道 Channel channel = connection.createChannel()){ // 绑定队列 参数第一个：队列名称 第二个：持久化 第三个:排他队列 第四个：自动删除 第五个：额外携带的参数 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //模拟生产者能生产很多消息 for (int i=0; i<20;i++){ // 消息内容 String message = "Hello World!"+i; //发送消息 第一个参数是交换机，第三个：对应的参数 第四个：消息的实体 channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8)); System.out.println(" [x] Sent '" + message + "'"+i); } } } }

创建两个消费者：

Recv01：

讯享网package com.xxxx.work.rr.recv; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DeliverCallback; /* * 工作队列（轮询）：消息的消费者 * * */ public class Recv01 { //定义队列名称 private final static String QUEUE_NAME = "work_rr"; public static void main(String[] argv) throws Exception { // 创建连接工厂 ConnectionFactory factory=new ConnectionFactory(); //相关配置 factory.setHost("192.168.23.129"); factory.setUsername("yeb"); factory.setVirtualHost("/yeb"); factory.setPassword(""); factory.setPort(5672); //连接工厂创建连接 Connection connection=factory.newConnection(); //创建信道 Channel channel = connection.createChannel(); //绑定队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); System.out.println("[*] Waitint for message .To exit press CTRL+C"); DeliverCallback deliverCallback=(consumerTag, delivery) -> { //睡2秒 模拟消费耗时 try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } String message=new String(delivery.getBody(),"UTF-8"); System.out.println(" [x] Received '" + message + "'"); //手动确认 第二个参数：是否确认多条 channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false); }; //监听队列消费消息 第二个参数：消息自动回值告诉生产者收到消息 channel.basicConsume(QUEUE_NAME,false,deliverCallback,consumerTag -> { }); } }

Recv02:

package com.xxxx.work.rr.recv; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DeliverCallback; /* * 工作队列（轮询）：消息的消费者 * * */ public class Recv02 { //定义队列名称 private final static String QUEUE_NAME = "work_rr"; public static void main(String[] argv) throws Exception { // 创建连接工厂 ConnectionFactory factory=new ConnectionFactory(); //相关配置 factory.setHost("192.168.23.129"); factory.setUsername("yeb"); factory.setVirtualHost("/yeb"); factory.setPassword(""); factory.setPort(5672); //连接工厂创建连接 Connection connection=factory.newConnection(); //创建信道 Channel channel = connection.createChannel(); //绑定队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); System.out.println("[*] Waitint for message .To exit press CTRL+C"); DeliverCallback deliverCallback=(consumerTag, delivery) -> { //睡2秒 模拟消费耗时 try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } String message=new String(delivery.getBody(),"UTF-8"); System.out.println(" [x] Received '" + message + "'"); //手动确认 第二个参数：是否确认多条 channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false); }; //监听队列消费消息 第二个参数：消息自动回值告诉生产者收到消息 channel.basicConsume(QUEUE_NAME,false,deliverCallback,consumerTag -> { }); } }

启动消费之Recv01、Recv02:

因为两个消费者，创建了2个工厂，所有有两个连接

每个连接创建一个信道： 2个信道

队列：

启动消息发送者发送消息：

消费者01：接收偶数

消费者02：接收基数

通过轮询，主要为了解决生产者的生产能力远远大于消费者的能力的时候，我们给它加多几个消费者，这就是轮询模式

（7）工作队列-公平

公平：解决了消费能力高低的问题：

工作队列轮询的优点，解决了简单队列的缺点，当我们的生产者生产能力远远大于消费者消费能力的时候，我给你加消费者，然后让消费者的消费能力能够大于等于生产者的生产能力，这样可以减少多余的消息堆积在消息队列里面，它也有缺点

比如说发送20条消息，如果使用轮询的话，没人消费一条分配，每个消费者消费10条，他们的消费能力不同 C1 1秒消费一条，C2 2秒消费一条消费完10条，这样C1用10秒，C2用20秒

有2个缺点：C1消费完10条之后要等C2，C1消费完之后要在这里等，资源就空到这里了，造成浪费，整体的消费时间也是比较长的，那么C1消费的快能不能帮助C2消费呢，C1帮C2没消费完的也进行消费，那么就用到公平模式，那么就用到公平模式，公平并不是生活中的一个一半，更意味更大的是能力越强，责任越大，

加啥上下面代码：

发送者：

讯享网package com.xxxx.work.fair.send; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.nio.charset.StandardCharsets; /* * 工作队列（公平）：消息的生产者 * * */ public class Send { //定义队列名称 private final static String QUEUE_NAME = "work_fair"; public static void main(String[] argv) throws Exception { // 创建连接工厂 ConnectionFactory factory=new ConnectionFactory(); //相关配置 factory.setHost("192.168.23.129"); factory.setUsername("yeb"); factory.setVirtualHost("/yeb"); factory.setPassword(""); factory.setPort(5672); try( //连接工厂创建连接 Connection connection=factory.newConnection(); //创建信道 Channel channel = connection.createChannel()){ // 绑定队列 参数第一个：队列名称 第二个：持久化 第三个:排他队列 第四个：自动删除 第五个：额外携带的参数 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //模拟生产者能生产很多消息 for (int i=0; i<20;i++){ // 消息内容 String message = "Hello World!"+i; //发送消息 第一个参数是交换机，第三个：对应的参数 第四个：消息的实体 channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8)); System.out.println(" [x] Sent '" + message + "'"+i); } } } }

消费者上面加入：

消费者1设置1秒消费一条消息：

package com.xxxx.work.fair.recv; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DeliverCallback; /* * 工作队列（公平）：消息的消费者 * * */ public class Recv01 { //定义队列名称 private final static String QUEUE_NAME = "work_fair"; public static void main(String[] argv) throws Exception { // 创建连接工厂 ConnectionFactory factory=new ConnectionFactory(); //相关配置 factory.setHost("192.168.23.129"); factory.setUsername("yeb"); factory.setVirtualHost("/yeb"); factory.setPassword(""); factory.setPort(5672); //连接工厂创建连接 Connection connection=factory.newConnection(); //创建信道 Channel channel = connection.createChannel(); //绑定队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); System.out.println("[*] Waitint for message .To exit press CTRL+C"); int prefetchCount=1; //限制消费者每次只能接收一条消息，处理完才能接收下一条消息 channel.basicQos(prefetchCount); DeliverCallback deliverCallback=(consumerTag, delivery) -> { //睡2秒 模拟消费耗时 try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } String message=new String(delivery.getBody(),"UTF-8"); System.out.println(" [x] Received '" + message + "'"); //手动确认 第二个参数：是否确认多条 channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false); }; //监听队列消费消息 第二个参数：消息自动回值告诉生产者收到消息 channel.basicConsume(QUEUE_NAME,false,deliverCallback,consumerTag -> { }); } }

消费者1设置2秒消费一条消息

讯享网package com.xxxx.work.fair.recv; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DeliverCallback; /* * 工作队列（公平）：消息的消费者 * * */ public class Recv02 { //定义队列名称 private final static String QUEUE_NAME = "work_fair"; public static void main(String[] argv) throws Exception { // 创建连接工厂 ConnectionFactory factory=new ConnectionFactory(); //相关配置 factory.setHost("192.168.23.129"); factory.setUsername("yeb"); factory.setVirtualHost("/yeb"); factory.setPassword(""); factory.setPort(5672); //连接工厂创建连接 Connection connection=factory.newConnection(); //创建信道 Channel channel = connection.createChannel(); //绑定队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); System.out.println("[*] Waitint for message .To exit press CTRL+C"); int prefetchCount=1; //限制消费者每次只能接收一条消息，处理完才能接收下一条消息 channel.basicQos(prefetchCount); DeliverCallback deliverCallback=(consumerTag, delivery) -> { //睡2秒 模拟消费耗时 try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } String message=new String(delivery.getBody(),"UTF-8"); System.out.println(" [x] Received '" + message + "'"); //手动确认 第二个参数：是否确认多条 channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false); }; //监听队列消费消息 第二个参数：消息自动回值告诉生产者收到消息 channel.basicConsume(QUEUE_NAME,false,deliverCallback,consumerTag -> { }); } }

启动2个消费者：

启动消息发送者：

消费者1：消费快，多消费了消息

消费者2：消费慢，少消费消息

公平模式，解决了不同消费者，不同的消费能力的差别的问题，做到能者多劳，能力强的多消费消息，能力落的少消费消息，没有资源浪费了解决了上面轮询消费者能力不同的问题，C1不会空在哪里，造成资源的浪费，让它一直干活，不会资源的浪费，第二整体的消费时间会变短。

（8）发布-订阅队列

公平队列，不能满足我们的需求，公平模式消费者接收的消息不能重复，如果我们有一个需求，我现在想要发送一条消息，然后被所有的消费者都能接收到，用我们上面所学的那两队列种模式无法实现，他们不能接收重复的消息，下面就用到第三种模式：

多了一个东西叫：交换机

交换机：交换机对每一个不同的虚拟路由，默认给了7个交换机

之前消息发送者直接把消息发送给队列，现在是生产者把消息，发送给交换机，交换机绑定不同的队列，然后把消息发送给队列，每个队列后面对应着一个消费者，他们去监听不同的队列，然后达到不同的消费者接收同一条消息的目的

消息订阅：广播类型：

消息创建者：

package com.xxxx.exchanges.send; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.nio.charset.StandardCharsets; /* * 发布/订阅队列：消息的生产者 * * */ public class Send { //定义交换机名称 private final static String EXCHANGE_NAME = "work_fanout"; public static void main(String[] argv) throws Exception { // 创建连接工厂 ConnectionFactory factory=new ConnectionFactory(); //相关配置 factory.setHost("192.168.23.129"); factory.setUsername("yeb"); factory.setVirtualHost("/yeb"); factory.setPassword(""); factory.setPort(5672); try( //连接工厂创建连接 Connection connection=factory.newConnection(); //创建信道 Channel channel = connection.createChannel()){ //绑定交换机 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT); // 消息内容 String message = "Hello World!"; //发送消息 第一个参数是交换机，第三个：对应的参数 第四个：消息的实体 channel.basicPublish(EXCHANGE_NAME,"", null, message.getBytes(StandardCharsets.UTF_8)); System.out.println(" [x] Sent '" + message + "'"); } } }

消费者：

讯享网package com.xxxx.exchanges.recv; import com.rabbitmq.client.*; /* * 发布/订阅队列：消息的消费者 * * */ public class Recv01 { //定义队列名称 private final static String EXCHANGE_NAME = "work_fanout"; public static void main(String[] argv) throws Exception { // 创建连接工厂 ConnectionFactory factory=new ConnectionFactory(); //相关配置 factory.setHost("192.168.23.129"); factory.setUsername("yeb"); factory.setVirtualHost("/yeb"); factory.setPassword(""); factory.setPort(5672); //连接工厂创建连接 Connection connection=factory.newConnection(); //创建信道 Channel channel = connection.createChannel(); //绑定交换机 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT); //获取队列（排他队列） String queueName=channel.queueDeclare().getQueue(); //绑定队列和交换机、 channel.queueBind(queueName,EXCHANGE_NAME,""); System.out.println("[*] Waitint for message .To exit press CTRL+C"); DeliverCallback deliverCallback=(consumerTag, delivery) -> { String message=new String(delivery.getBody(),"UTF-8"); System.out.println(" [x] Received '" + message + "'"); }; //监听队列消费消息 第二个参数：消息自动回值告诉生产者收到消息 channel.basicConsume(queueName,true,deliverCallback,consumerTag -> { }); } }

package com.xxxx.exchanges.recv; import com.rabbitmq.client.*; /* * 发布/订阅队列：消息的消费者 * * */ public class Recv02 { //定义队列名称 private final static String EXCHANGE_NAME = "work_fanout"; public static void main(String[] argv) throws Exception { // 创建连接工厂 ConnectionFactory factory=new ConnectionFactory(); //相关配置 factory.setHost("192.168.23.129"); factory.setUsername("yeb"); factory.setVirtualHost("/yeb"); factory.setPassword(""); factory.setPort(5672); //连接工厂创建连接 Connection connection=factory.newConnection(); //创建信道 Channel channel = connection.createChannel(); //绑定交换机 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT); //获取队列（排他队列） String queueName=channel.queueDeclare().getQueue(); //绑定队列和交换机 channel.queueBind(queueName,EXCHANGE_NAME,""); System.out.println("[*] Waitint for message .To exit press CTRL+C"); DeliverCallback deliverCallback=(consumerTag, delivery) -> { String message=new String(delivery.getBody(),"UTF-8"); System.out.println(" [x] Received '" + message + "'"); }; //监听队列消费消息 第二个参数：消息自动回值告诉生产者收到消息 channel.basicConsume(queueName,true,deliverCallback,consumerTag -> { }); } }

运行消费01、02：