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

RabbitMQ初识

news 2025/1/21 6:37:55

Kafka

RocketMQ

RabbitMQ

MQ界面(它使用的端口号5672，界面是15672）

如何添加一个虚拟机，点击右侧

Topics（通配符模式）

发布确认机制

持久性(可靠性保证的机制之一)

JDK17,Linux服务器Ubuntu

什么是MQ

实现了AMQP消息队列服务，当前主流消息中间件一。

AMQP:高级消息队列协议，是一个通用的应用层协议，通过统一消息服务的协议，为面向消息中间件设计，基于此协议的客户端与消息中间件可传递消息

MQ消息队列:

先进先出

同步通信:

A->B

异步通信:

A->消息队列->B 消息传递

MQ的作用:接收并且转发消息

1.异步解耦:

用户注册:

1)用户信息校验

2)插入数据库

3)发送邮件

生活中会有一些操作非常耗时，但是并不需要即使返回结果，可以借助MQ把这些操作异步化，比如用户注册之后，发送注册成功，可以作为异步任务处理，而不必等待这些操作完成之后，才告诉用户注册成功

2.流量削峰:在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见，如果处理这类峰值为标准投入资源，肯定是浪费，所以关键时刻，使用MQ使关键组件支撑突发访问压力，不会因为突发流量而崩溃，比如xx关宣恋情，省钱之眼秒杀活动，可以使用MQ控制流量，请求排队，然后再系统根据自己能力逐渐处理这些请求

3.消息分发:当多个系统需要对同一数据做出相应时，可以使用MQ进行消息分发，比如支付成功之后，支付系统可以像MQ发消息，其他系统订阅该消息，这样就不需要轮询数据库了

每月付款

4.延时通知:在需要特定时间后发送通知的场景中，可以使用MQ的延迟消息功能，比如在电子商务平台中，如果用户下单后一定时间内未支付，可以使用延时消息队列在超时后自动取消订单。

Kafka

开始运用的目的是日志的手机和传输，追求高吞吐，性能卓越，单机吞吐十万级，但是功能比较简单，日志领域倒是比较成熟

RocketMQ

RocketMQ采用JAVA开发，由阿里巴巴开源，后捐赠给Apache

他设计时借鉴了Kafka，并做出改进，经多年双十一系例，适用于可靠性比较高，并且并发比较大的场景，比如互联网金融，但是支持的客户端语言不多，并且社区活跃一般（意味着遇到问题的话，没有很多的解决方案)

RabbitMQ

采用Erlang语言，MQ功能完善，并且几乎支持所有主流语言，开源界面也十分友好，性能比较好，吞吐量达到数万级别，社区活跃度也高，适合中小型公司，数据量没那么大，并发没那么高的场景

RabbitMQ安装需要Erlang语言的支持，在安装rabbitMq之前需要安装erlang

至于如何下载rabbitMq,看我前面docker的操作，之间拉镜像，然后看你需不需要保存本地

docker run -d  --name rabbitmq -p 15672:15672 -p 5672:5672 rabbitmq
docker exec -it 镜像名字 bash
rabbitmq-plugins enable rabbitmq_management

MQ界面(它使用的端口号5672，界面是15672）

虚拟机，类似于mysql的database。

添加一个角色，那个黄色代表咩有任何权限，所以

点击进去后set就好了。

如何添加一个虚拟机，点击右侧

添加成功，guest有权限

AMQP:高级消息队列协议

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。IP 2.端口号 3.账号 4.密码 5。虚拟机//1.建立连接ConnectionFactory connectionFactory=new ConnectionFactory();connectionFactory.setHost("47.98.61.112");connectionFactory.setPort(5672);connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");connectionFactory.setVirtualHost("bite");Connection connection=connectionFactory.newConnection();//2.开启信道Channel channel= connection.createChannel();//3.声明交换机,使用内置交换机/*** queueDeclare(String queue,boolean durable,boolean exclusive,boolean outoDelete,Map<String,Object>arguments)* 参数说明* queue:队列名称* durable:可持久化* exclusive:是否独占* autoDelete:是否自动删除* arguments:参数*///4.声明队列channel.queueDeclare("hello",true,false,false,null);//5.发送消息/*** basicPublish(String exchange,String routingKey,BasicProperties props,byte[]body)* 参数说明:* exchange:交换机名称* routingKey:内置交换机,routingKey和队列名称保持一致* props:属性配置* body：消息*/String msq="hello rabbitmq~";channel.basicPublish("","hello",null,msq.getBytes());System.out.println("我喜欢消息成功");//6.资源释放channel.close();connection.close();}
}

我们可以看到图中发送成功

假如我们不去释放资源，那么他就不结束

那么此处他就还在连接

消费者代码与生产者代码相似

1.创建链接

2.创建Channel

3.声明一个队列Queue

4.消费消息

5.释放资源

生产者不生产，请问我怎么消费捏？

为什么消费者需要声明队列:

消费者启动时候，需要制定订阅的队列，如果当时队列不存在，消费者会报错

消费者接受消息

import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class ConsumerDemo {/*** 1.创建连接* 2.创建Channel* 3.声明队列(可以省略,前提生产者声明好了，生产者要在消费者启动之前，声明好，最好声明)* 4。消费消息* 5。释放资源*/public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {ConnectionFactory connectionFactory=new ConnectionFactory();connectionFactory.setHost("47.98.61.112");connectionFactory.setPort(5672);connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");connectionFactory.setVirtualHost("bite");Connection connection=connectionFactory.newConnection();//2.创建ChannelChannel 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){//从队列中收到消息，就会执行的方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//TODOSystem.out.println("接收到消息"+new String(body));}};channel.basicConsume("hello",true,consumer);Thread.sleep(2000);//5。释放资源channel.close();connection.close();}}

如果我们不去声明队列，那么就会报下面的错误，如果声明了，就算你把之前的队列删除，他会再给你声明。

RabbitMQ工作模式

上面两个只是起到了消息转发，所以没画，但是rabbitMQ不用上面这两个模式，其余的用

发布订阅模式，对应的是Fanout广播/扇出？

RoutingKey:发送的话指定 BindingKey:是RoutingKey的一种，是一种用来绑定的RoutingKey（路由键:生产者将消息发给交换器时，指定的一个字符串，用来告诉交换机应该如何处理这个消息)

Direct:定向,把消息交给指定routingkey的队列(Routing模式)

Topic通配符:把消息交给符合routing Pattermn路由模式的队列

使用绑定的时候，需要的路由键时BindingKey

发送消息的时候，需要的路由键时RoutingKey

4.路由模式(发布订阅模式的变种):

Topics（通配符模式）

RPC通信

发布确认机制

RabbitMQ消息可靠性的机制:他只是保证他可以正确发送到rabbitmq,生产者将Channel设置为Confirm模式(通过调用,channel.confirmSelect完成后，发布的每一条消息，都会获得一个唯一的ID,生产者可以将这些序列号与消息关联起来，以便于跟踪消息的状态

2.当消息被RabbitMq服务器接受并且处理之后，服务器会异步的向生产者发送一个确认(ACK)给生产者(包含消息的唯一iD),表明消息已经送达。

AcknowledgeMode.NONE:

这种模式下，消息一旦投递给消费者，不管消费者是否成功处理了消息，RabbitMQ就会自动确认消息，从RabbitMQ队列中移除消息，如果消费者处理消息失败，消息可能会丢失.

消费者正常处理:MQ删除相应消息

消费者异常处理:MQ删除相应消息

AcknowledgeMode.AUTO(默认)

这种模式下，消费者在消息处理成功时候，会自动确认消息，但是如果处理过程抛出异常，则不会确认消息

消费者正常处理:MQ正常确认

消费者异常处理:弹出一直弹错误(因为他会一直重新发消息)

AcknowledgeMode.MANUAL

手动确认模式下,消费者必须在成功处理消息后显式调用basicAck方法来确认消息，如果消息未被确认,RabbitMQ会认为消息尚未被成功处理，并且会在消费者可用时重新投递该消息，这种模式提高了消息处理的可靠性，因为即使消费者处理消息后失效，消息也不会丢失，而是可以被重新处理。

持久性(可靠性保证的机制之一)

RabbitMQ退出，由于某种原因崩溃时候，会忽视队列和消息，所以这也就需要持久化来帮助

RabbitMQ持久化包括三个部分，交换器持久化，队列的持久化，消息的持久化

交换机持久化是通过声明交换机，将durable设置为true,这可以理解为存硬盘上了,相当于将交换机的属性在服务器内部保存，当MQ的服务器发生意外或者关闭后，RabbitMQ不需要重新建立交换机，交换机会自动建立，相当于一直存在。

默认是持久化。

这样就是非持久化

持久化队列

非持久化队列

消息存储在队列中

消息持久化,需要队列持久化+消息持久化

只设置了队列持久化，MQ重启后消息会丢失

只设置消息持久化,MQ重启后，队列会丢失。消息也会丢失。

消息持久化的源码是判断是否是Message对象，假如是这个对象，就直接原封不动的返回.

测试场景

1.交换机持久化 -数据不丢失

非持久化 -数据丢失(交换机都丢了)

队列持久化 -依然存在不丢失

消息持久化 -消息存在，没有丢失

消息非持久化 -消息丢失，队列没丢。

队列非持久化 -队列都丢失了，你也看不到消息了

消息持久化 -消息丢失，队列还在。

消息非持久化 -队列也丢失了，消息也丢失

在持久化消息正确存入RabbitMQ之后，还有一段时间，（虽然很短，但是也不能忽视）才能存入磁盘中，RabbitMQ并不会每条消息都进行同步存盘的处理，可能仅仅保存到操作系统缓存之中而不是物理磁盘之中，如果这段时间内，RabbitMQ服务节点发生了宕机或者重启等异常情况，消息保存，还没来得及罗盘，那么这些消息将会丢失。

引入RabbitMQ的仲裁队列，假如主节点在此特殊节点挂掉，可以自动切换从节点，有效的保证高可用性，除非整个集群都挂掉(此方法也不保证100%可靠,但是配置了仲裁队列要比没有配置仲裁队列的可靠性要高很多，实际生产环境中的关键业务队列一般都会设置仲裁队列

2.还可以在发送端引入事务机制或者发送方确认机制，保证消息已经正确的发送并且存储至RabbitMQ。