RabbitMQ初识

server/2024/10/18 10:28:12/

目录

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。

小知识

int范围-127 -127 存储放到栈里面存储,超过200,地址可能有问题

springboot项目启动要注意,把那个该死的application放到最外面,不然无法找到里面的Mapping。


http://www.ppmy.cn/server/131431.html

相关文章

RabbitMQ 交换机的类型

在 RabbitMQ 中&#xff0c;交换机&#xff08;Exchange&#xff09;是一个核心组件&#xff0c;负责接收来自生产者的消息&#xff0c;并根据特定的路由规则将消息分发到相应的队列。交换机的存在改变了消息发送的模式&#xff0c;使得消息的路由更加灵活和高效。 交换机的类…

react-JSX

JSX理念 jsx在编译的时候会被babel编译为react.createELement方法 在使用jsx的文件中&#xff0c;需要引入react。import React from "react" jsx会被编译为React.createElement,所有jsx的运行结果都是react element React Component 在react中&#xff0c;常使用…

k8s的安装和部署

配置三台主机&#xff0c;分别禁用各个主机上的swap&#xff0c;并配置解析 systemctl mask swap.target swapoff -a vim /etc/fstab配置这三个主机上的主机以及harbor仓库的主机 所有主机设置docker的资源管理模式为system [rootk8s-master ~]# vim /etc/docker/daemon.json…

238. 除自身以外数组的乘积

文章目录 1.题目2.思路3.代码 1.题目 238. 除自身以外数组的乘积 给你一个整数数组 nums&#xff0c;返回 数组 answer &#xff0c;其中 answer[i] 等于 nums 中除 nums[i] 之外其余各元素的乘积 。 题目数据 保证 数组 nums之中任意元素的全部前缀元素和后缀的乘积都在 32…

【WEB应用安全测试指南–蓝队安全测试2】--超详细-可直接进行实战!!!亲测-可进行安全及渗透测试

安全基础理论入门知识参考上一篇《WEB应用安全测试指南蓝队安全测试1》 WEB应用安全测试指南2 一、文件 I/O 类1.1、任意文件上传1.2、任意文件下载1.3、文件包含 二、接口安全类2.1、短信炸弹2.2、邮件炸弹2.3、短信内容可控2.4、邮件内容可控 三、逻辑流程类3.1、越权3.2、未…

大数据Lambda 架构和Kappa 架构的优缺点和使用场景

Lambda 架构和Kappa 架构是用于处理大数据的两种架构设计模式&#xff0c;针对不同的数据处理需求提供了不同的方案。它们各自有优缺点&#xff0c;并适用于不同的使用场景。 Lambda 架构 Lambda 架构分为三个层次&#xff1a;批处理层&#xff08;Batch Layer&#xff09;、…

分布式储能监控在实际项目中的应用

项目概述 晋亿实业股份有限公司位于浙江省嘉兴市嘉善县惠民街道松海路66号厂区&#xff0c;属于220kV东云变供区&#xff0c;由东云变35kV晋流683线供电。 晋亿实业5MW分布式储能项目&#xff08;以下简称“本项目”&#xff09;是某新能源科技有限公司在晋亿实业厂区内空地新…

大模型 memory 记忆 缓存的应用

在探讨大模型的“memory”&#xff08;记忆&#xff09;功能时&#xff0c;我们通常会涉及缓存、存储以及如何有效管理和利用这些记忆来增强模型的性能。以下是对大模型memory记忆、缓存及相关概念的详细分析&#xff1a; 一、大模型的记忆功能 大模型&#xff0c;特别是大型…