商城-学习整理-高级-消息队列(十七)

news2024/11/24 20:12:42

目录

  • 一、RabbitMQ简介(消息中间件)
    • 1、RabbitMQ简介:
    • 2、核心概念
      • 1、Message
      • 2、Publisher
      • 3、Exchange
      • 4、Queue
      • 5、Binding
      • 6、Connection
      • 7、Channel
      • 8、Consumer
      • 9、Virtual Host
      • 10、Broker
  • 二、一些概念
    • 1、异步处理
    • 2、应用解耦
    • 3、流量控制
    • 5、概述
  • 三、Docker安装RabbitMQ
  • 四、RabbitMQ运行机制
    • 1、AMQP 中的消息路由
    • 2、Exchange 类型
  • 五、RabbitMQ整合
  • 六、RabbitMQ消息确认机制-可靠抵达
    • 1、可靠抵达-ConfirmCallback
    • 2、可靠抵达-ReturnCallback
    • 3、可靠抵达-Ack消息确认机制
  • 七、RabbitMQ延时队列(实现定时任务)
    • 1、场景:
    • 2、消息的TTL(Time To Live)
    • 3、Dead Letter Exchanges(DLX)
    • 4、延时队列实现-1
    • 5、延时队列实现-2
    • 6、SpringBoot中使用延时队列
    • 7、如何保证消息可靠性-消息丢失
    • 8、如何保证消息可靠性-消息重复
    • 9、如何保证消息可靠性-消息积压
    • 10、MQ对比

一、RabbitMQ简介(消息中间件)

1、RabbitMQ简介:

RabbitMQ是一个由erlang开发的AMQP(Advanved Message Queue Protocol)的开源实现。

2、核心概念

1、Message

消息,消息是不具名的,它由消息头和消息体组成。消息体是不透明的,而消息头则由一系列的可选属性组成,
这些属性包括routing-key(路由键)、priority(相对于其他消息的优先权)、delivery-mode(指出该消息可
能需要持久性存储)等。

2、Publisher

消息的生产者,也是一个向交换器发布消息的客户端应用程序。

3、Exchange

交换器,用来接收生产者发送的消息并将这些消息路由给服务器中的队列。
Exchange有4种类型:direct(默认),fanout, topic, 和headers,不同类型的Exchange转发消息的策略有所区别

4、Queue

消息队列,用来保存消息直到发送给消费者。它是消息的容器,也是消息的终点。一个消息可投入一个或多个队列。消息一直
在队列里面,等待消费者连接到这个队列将其取走。

5、Binding

绑定,用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则,所以可以将交
换器理解成一个由绑定构成的路由表。
Exchange 和Queue的绑定可以是多对多的关系。

6、Connection

网络连接,比如一个TCP连接。

7、Channel

信道,多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内的虚拟连接,AMQP 命令都是通过信道
发出去的,不管是发布消息、订阅队列还是接收消息,这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都
是非常昂贵的开销,所以引入了信道的概念,以复用一条 TCP 连接。

8、Consumer

消息的消费者,表示一个从消息队列中取得消息的客户端应用程序。

9、Virtual Host

虚拟主机,表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器,拥有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础,必须在连接时指定,RabbitMQ 默认的 vhost 是 / 。

10、Broker

表示消息队列服务器实体
在这里插入图片描述

在这里插入图片描述

二、一些概念

1、异步处理

在这里插入图片描述

2、应用解耦

在这里插入图片描述

3、流量控制

在这里插入图片描述

5、概述

  1. 大多应用中,可通过消息服务中间件来提升系统异步通信、扩展解耦能力
  2. 消息服务中两个重要概念:
    消息代理(message broker)和目的地(destination)
    当消息发送者发送消息以后,将由消息代理接管,消息代理保证消息传递到指定目的地。
  3. 消息队列主要有两种形式的目的地
    1、 队列(queue):点对点消息通信(point-to-point)
    2、主题(topic):发布(publish)/订阅(subscribe)消息通信
  4. 点对点式:
    • 消息发送者发送消息,消息代理将其放入一个队列中,消息接收者从队列中获取消息内容,消息读取后被移出队列
    • 消息只有唯一的发送者和接受者,但并不是说只能有一个接收者
  5. 发布订阅式:
    • 发送者(发布者)发送消息到主题,多个接收者(订阅者)监听(订阅)这个主题,那么就会在消息到达时同时收到消息
  6. JMS(Java Message Service)JAVA消息服务:
    • 基于JVM消息代理的规范。ActiveMQ、HornetMQ是JMS实现
  7. AMQP(Advanced Message Queuing Protocol)
    • 高级消息队列协议,也是一个消息代理的规范,兼容JMS
    • RabbitMQ是AMQP的实现
    在这里插入图片描述
  8. Spring支持
    • spring-jms提供了对JMS的支持
    • spring-rabbit提供了对AMQP的支持
    • 需要ConnectionFactory的实现来连接消息代理
    • 提供JmsTemplate、RabbitTemplate来发送消息
    • @JmsListener(JMS)、@RabbitListener(AMQP)注解在方法上监听消息
    代理发布的消息
    • @EnableJms、@EnableRabbit开启支持
  9. Spring Boot自动配置
    • JmsAutoConfiguration
    • RabbitAutoConfiguration
    • 10、市面的MQ产品
    • ActiveMQ、RabbitMQ、RocketMQ、Kafka

三、Docker安装RabbitMQ

docker run -d --name rabbitmq -p 5671:5671 -p 5672:5672 -p 4369:4369 -p 25672:25672 -p 15671:15671 -p 15672:15672  rabbitmq:management

4369, 25672 (Erlang发现&集群端口)
5672, 5671 (AMQP端口)
15672 (web管理后台端口)
61613, 61614 (STOMP协议端口)
1883, 8883 (MQTT协议端口)
https://www.rabbitmq.com/networking.html

在这里插入图片描述

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

默认账号和密码是:guest/guest

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

四、RabbitMQ运行机制

1、AMQP 中的消息路由

• AMQP 中消息的路由过程和 Java 开发者熟悉的 JMS 存在一些差别,AMQP 中增加了 Exchange 和Binding 的角色。生产者把消息发布到 Exchange 上,消息最终到达队列并被消费者接收,而 Binding 决定交换器的消息应该发送到那个队列。
在这里插入图片描述

2、Exchange 类型

• Exchange分发消息时根据类型的不同分发策略有区别,目前共四种类型:direct、fanout、topic、headers 。headers 匹配 AMQP 消息的 header 而不是路由键,headers 交换器和 direct 交换器完全一致,但性能差很多,目前几乎用不到了,所以直接看另外三种类

消息中的路由键(routing key)如果和Binding 中的 binding key 一致, 交换器就将消息发到对应的队列中。路由键与队列名完全匹配,如果一个队列绑定到交换机要求路由键为“dog”,则只转发 routingkey 标记为“dog”的消息,不会转发“dog.puppy”,也不会转发“dog.guard” 等等。它是完全匹配、单播的模式。

在这里插入图片描述

每个发到 fanout 类型交换器的消息都会分到所有绑定的队列上去。fanout 交换器不处理路由键,只是简单的将队列绑定到交换器上,每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上。很像子网广播,每台子网内的主机都获得了一份复制的消息。fanout 类型转发消息是最快的。

在这里插入图片描述

topic 交换器通过模式匹配分配消息的路由键属性,将路由键和某个模式进行匹配,此时队列需要绑定到一个模式上。它将路由键和绑定键的字符串切分成单词,这些单词之间用点隔开。它同样也会识别两个通配符:符号“#”和符号“*”。#匹配0个或多个单词,*匹配一个单词。

在这里插入图片描述

五、RabbitMQ整合

  1. 引入 spring-boot-starter-amqp
    在这里插入图片描述

在这里插入图片描述

  1. application.yml配置
    在这里插入图片描述

  2. 测试RabbitMQ

  3. AmqpAdmin:管理组件

  4. RabbitTemplate:消息发送处理组件

  5. @RabbitListener 监听消息的方法可以有三种参数(不分数量,顺序)
    • Object content, Message message, Channel channel

六、RabbitMQ消息确认机制-可靠抵达

• 保证消息不丢失,可靠抵达,可以使用事务消息,性能下降250倍,为此引入确认机制
• publisher confirmCallback 确认模式
• publisher returnCallback 未投递到 queue 退回模式
• consumer ack机制
在这里插入图片描述

1、可靠抵达-ConfirmCallback

• spring.rabbitmq.publisher-confirms=true
• 在创建 connectionFactory 的时候设置 PublisherConfirms(true) 选项,开启confirmcallback 。
• CorrelationData:用来表示当前消息唯一性。
• 消息只要被 broker 接收到就会执行 confirmCallback,如果是 cluster 模式,需要所有
broker 接收到才会调用 confirmCallback。
• 被 broker 接收到只能表示 message 已经到达服务器,并不能保证消息一定会被投递到目标 queue 里。所以需要用到接下来的returnCallback 。

2、可靠抵达-ReturnCallback

• spring.rabbitmq.publisher-returns=true
• spring.rabbitmq.template.mandatory=true
• confrim 模式只能保证消息到达 broker,不能保证消息准确投递到目标 queue 里。在有些业务场景下,我们需要保证消息一定要投递到目标 queue 里,此时就需要用到return 退回模式。
• 这样如果未能投递到目标 queue 里将调用 returnCallback ,可以记录下详细到投递数据,定期的巡检或者自动纠错都需要这些数据。

3、可靠抵达-Ack消息确认机制

•消费者获取到消息,成功处理,可以回复Ack给Broker
•basic.ack用于肯定确认;broker将移除此消息
•basic.nack用于否定确认;可以指定broker是否丢弃此消息,可以批量
•basic.reject用于否定确认;同上,但不能批量
•默认自动ack,消息被消费者收到,就会从broker的queue中移除
•queue无消费者,消息依然会被存储,直到消费者消费
•消费者收到消息,默认会自动ack。但是如果无法确定此消息是否被处理完成,或者成功处理。我们可以开启手动ack模式
•消息处理成功,ack(),接受下一个消息,此消息broker就会移除
•消息处理失败,nack()/reject(),重新发送给其他人进行处理,或者容错处理后ack
•消息一直没有调用ack/nack方法,broker认为此消息正在被处理,不会投递给别人,此时客户端断开,消息不会被broker移除,会投递给别人

七、RabbitMQ延时队列(实现定时任务)

1、场景:

比如未付款订单,超过一定时间后,系统自动取消订单并释放占有物品。
常用解决方案:
spring的 schedule 定时任务轮询数据库
缺点:
消耗系统内存、增加了数据库的压力、存在较大的时间误差
解决:rabbitmq的消息TTL和死信Exchange结合

2、消息的TTL(Time To Live)

• 消息的TTL就是消息的存活时间。
• RabbitMQ可以对队列和消息分别设置TTL。
• 对队列设置就是队列没有消费者连着的保留时间,也可以对每一个单独的消息做单独的设置。超过了这个时间,我们认为这个消息就死了,称之为死信。
• 如果队列设置了,消息也设置了,那么会取小的。所以一个消息如果被路由到不同的队列中,这个消息死亡的时间有可能不一样(不同的队列设置)。这里单讲单个消息的TTL,因为它才是实现延迟任务的关键。可以通过设置消息的expiration字段或者x- message-ttl属性来设置时间,两者是一样的效果。

3、Dead Letter Exchanges(DLX)

• 一个消息在满足如下条件下,会进死信路由,记住这里是路由而不是队列,一个路由可以对应很多队列。(什么是死信)
• 一个消息被Consumer拒收了,并且reject方法的参数里requeue是false。也就是说不会被再次放在队列里,被其他消费者使用。(basic.reject/ basic.nack)requeue=false
• 上面的消息的TTL到了,消息过期了。
• 队列的长度限制满了。排在前面的消息会被丢弃或者扔到死信路由上
• Dead Letter Exchange其实就是一种普通的exchange,和创建其他
exchange没有两样。只是在某一个设置Dead Letter Exchange的队列中有消息过期了,会自动触发消息的转发,发送到Dead Letter Exchange中去。
• 我们既可以控制消息在一段时间后变成死信,又可以控制变成死信的消息被路由到某一个指定的交换机,结合二者,其实就可以实现一个延时队列
• 手动ack&异常消息统一放在一个队列处理建议的两种方式
• catch异常后,手动发送到指定队列,然后使用channel给rabbitmq确认消息已消费
• 给Queue绑定死信队列,使用nack(requque为false)确认消息消费失败

在这里插入图片描述

4、延时队列实现-1

在这里插入图片描述

5、延时队列实现-2

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

6、SpringBoot中使用延时队列

• 1、Queue、Exchange、Binding可以@Bean进去
• 2、监听消息的方法可以有三种参数(不分数量,顺序)
• Object content, Message message, Channel channel
• 3、channel可以用来拒绝消息,否则自动ack;

7、如何保证消息可靠性-消息丢失

• 1、消息丢失
• 消息发送出去,由于网络问题没有抵达服务器
• 做好容错方法(try-catch),发送消息可能会网络失败,失败后要有重试机制,可记录到数据库,采用定期扫描重发的方式
• 做好日志记录,每个消息状态是否都被服务器收到都应该记录
• 做好定期重发,如果消息没有发送成功,定期去数据库扫描未成功的消息进行重发
• 消息抵达Broker,Broker要将消息写入磁盘(持久化)才算成功。此时Broker尚未持久化完成,宕机。
• publisher也必须加入确认回调机制,确认成功的消息,修改数据库消息状态。
• 自动ACK的状态下。消费者收到消息,但没来得及消息然后宕机
• 一定开启手动ACK,消费成功才移除,失败或者没来得及处理就noAck并重新入队

8、如何保证消息可靠性-消息重复

2、消息重复
• 消息消费成功,事务已经提交,ack时,机器宕机。导致没有ack成功,Broker的消息重新由unack变为ready,并发送给其他消费者
• 消息消费失败,由于重试机制,自动又将消息发送出去
• 成功消费,ack时宕机,消息由unack变为ready,Broker又重新发送
• 消费者的业务消费接口应该设计为幂等性的。比如扣库存有工作单的状态标志
• 使用防重表(redis/mysql),发送消息每一个都有业务的唯一标识,处理过就不用处理
• rabbitMQ的每一个消息都有redelivered字段,可以获取是否是被重新投递过来的,而不是第一次投递过来的

9、如何保证消息可靠性-消息积压

3、消息积压
• 消费者宕机积压
• 消费者消费能力不足积压
• 发送者发送流量太大
• 上线更多的消费者,进行正常消费
• 上线专门的队列消费服务,将消息先批量取出来,记录数据库,离线慢慢处理

10、MQ对比

ActiveMQ、RabbitMQ、RocketMQ、Kafka、Flume

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/909547.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

开源全球地理空间数据可视化框架——Cesium学习(2023.8.21)

Cesium学习 2023.8.21 1、Cesium简介1.1 Github上的Cesium 2、Cesium下载安装使用2.1 方式一:页面在线引用2.2 方式二:页面离线使用2.3 方式三:完整项目使用 3、CesiumJS学习教程(快速上手 API文档)3、Cesium官方示例…

EasyCode代码生成MybatisPlus

先安装插件 导入 { "author" : "wangyujie", "version" : "1.2.8", "userSecure" : "", "currTypeMapperGroupName" : "Default", "currTemplateGroupName" : "01-Mybatis-Pl…

HTML a标签

<a>标签定义一个超链接。它有如下主要属性&#xff1a; href&#xff1a;指定链接的地址&#xff0c;可以是一个URL、文件路径或锚点。target&#xff1a;指定链接在何处打开。其值包括&#xff1a; _blank&#xff1a;在新窗口或新标签页打开链接。_self&#xff1a;在…

【笔记】Spark3 AQE(Adaptive Query Execution)

提效 7 倍&#xff0c;Apache Spark 自适应查询优化在网易的深度实践及改进 Performance Tuning 配置Spark SQL开启Adaptive Execution特性 How To Use Spark Adaptive Query Execution (AQE) in Kyuubi 【spark系列3】spark 3.0.1 AQE(Adaptive Query Exection)分析 玩转Spark…

jmeter入门:接口压力测试全解析

一.对接口压力测试 1.配置 1.添加线程组&#xff08;参数上文有解释 这里不介绍&#xff09; 2.添加取样器 不用解释一看就知道填什么。。。 3.添加头信息&#xff08;否则请求头对不上&#xff09; 也不用解释。。。 4.配置监听器 可以尝试使用这几个监听器。 2.聚合结果…

MySQL 字符集概念、原理及如何配置 — 图文详解

目录 一、字符集概念 1、字符&#xff08;Character&#xff09; 2、字符编码 3、字符集&#xff08;Character set&#xff09; 二、字符集原理 1. ASCII字符集 2、GB2312 3、GBK 4、GB18030 5、BIG5 6、Unicode 编码 三、字符序 四、MySQL字符集 & 字符序 …

web基础+HTTP协议+httpd详细配置

目目录录 一、Web基础1.1 HTML概述1.1.1 HTML的文件结构1.1.2 HTML中的部分基本标签 1.3 MIME1.4 URI 和 URL1.4 定义1.4.2 URI 和 URL 的区别 二、静态资源和动态资源2.1 静态资源2.2 动态资源 三、HTTP协议3.1 HTTP协议简介3.2 HTTP协议版本3.2 HTTP方法3.3 HTTP请求访问的完…

李宏毅机器学习笔记:结构学习,HMM,CRF

李宏毅机器学习笔记&#xff1a;结构学习&#xff0c;HMM&#xff0c;CRF 1、隐马尔可夫模型HMM1.1Sequence2Sequence1.2 HMM1.3 Viterbi算法1.3 HMM模型的缺点 1、隐马尔可夫模型HMM 1.1Sequence2Sequence 什么是Seq2Seq问题呢&#xff1f;简单来说&#xff0c;就是输入是一…

RTSP/Onvif视频服务器EasyNVR安防视频云服务调用接口录像会被自动删除的问题解决方案

EasyNVR安防视频云服务是基于RTSP/Onvif协议接入的视频平台&#xff0c;可支持将接入的视频流进行全平台、全终端的分发&#xff0c;分发的视频流包括RTSP、RTMP、HTTP-FLV、WS-FLV、HLS、WebRTC等。平台丰富灵活的视频能力&#xff0c;可应用在智慧校园、智慧工厂、智慧水利等…

Python“牵手”京东商品详情数据采集方法,京东API申请步骤说明

京东平台API接口是为开发电商类应用程序而设计的一套完整的、跨浏览器、跨平台的接口规范。 京东API接口是指通过编程的方式&#xff0c;让开发者能够通过HTTP协议直接访问京东平台的数据&#xff0c;包括商品信息、店铺信息、物流信息等&#xff0c;从而实现京东平台的数据开…

自动驾驶合成数据科普一:不做真实数据的“颠覆者”,做“杠杆”

前言&#xff1a; 在7月底的一篇文章中&#xff0c;九章智驾提到&#xff0c;数据闭环能力是自动驾驶下半场的“入场券”&#xff0c;这一观点在行业内引起了广泛共鸣。 在数据闭环体系中&#xff0c;仿真技术无疑是非常关键的一环。仿真的起点是数据&#xff0c;而数据又分为真…

回归预测 | MATLAB实现WOA-RBF鲸鱼优化算法优化径向基函数神经网络多输入单输出回归预测(多指标,多图)

回归预测 | MATLAB实现WOA-RBF鲸鱼优化算法优化径向基函数神经网络多输入单输出回归预测&#xff08;多指标&#xff0c;多图&#xff09; 目录 回归预测 | MATLAB实现WOA-RBF鲸鱼优化算法优化径向基函数神经网络多输入单输出回归预测&#xff08;多指标&#xff0c;多图&#…

短视频seo源码矩阵系统开源---代码php分享

前言&#xff1a;短视频seo源码 短视频seo矩阵系统源码私有化部署 短视频seo源码 短视频seo矩阵系统源码私有化怎么部署&#xff1f; 首先我们来给大家普及一下什么是短视频seo矩阵系统&#xff1f;视频矩阵分为多平台矩阵与一个平台多账号矩阵&#xff0c;加上seo排名优化&…

EasyPOI 实战总结

EasyPOI实战总结 简介 easypoi功能如同名字easy,主打的功能就是容易,让一个没见接触过poi的人员 就可以方便的写出Excel导出,Excel模板导出,Excel导入,Word模板导出,通过简单的注解和模板 语言(熟悉的表达式语法),完成以前复杂的写法 使用EasyPOI 环境搭建 # 1.引入相关依…

R包开发一:R与Git版本控制

目录 1.安装Git 2-配置Git&#xff08;只需配置一次&#xff09; 3-用SSH连接GitHub(只需配置一次) 4-创建Github远程仓库 5-克隆仓库到本地 目标&#xff1a;创建的R包&#xff0c;包含Git版本控制&#xff0c;并且能在远程Github仓库同步&#xff0c;相当于发布在Github。…

基于Spring Boot的游泳馆管理系统的设计与实现(Java+spring boot+MySQL)

获取源码或者论文请私信博主 演示视频&#xff1a; 基于Spring Boot的游泳馆管理系统的设计与实现&#xff08;Javaspring bootMySQL&#xff09; 使用技术&#xff1a; 前端&#xff1a;html css javascript jQuery ajax thymeleaf 微信小程序 后端&#xff1a;Java spring…

R语言机器学习方法在生态经济学领域

近年来&#xff0c;人工智能领域已经取得突破性进展&#xff0c;对经济社会各个领域都产生了重大影响&#xff0c;结合了统计学、数据科学和计算机科学的机器学习是人工智能的主流方向之一&#xff0c;目前也在飞快的融入计量经济学研究。表面上机器学习通常使用大数据&#xf…

HarmonyOS学习路之方舟开发框架—学习ArkTS语言(状态管理 四)

Observed装饰器和ObjectLink装饰器&#xff1a;嵌套类对象属性变化 上文所述的装饰器仅能观察到第一层的变化&#xff0c;但是在实际应用开发中&#xff0c;应用会根据开发需要&#xff0c;封装自己的数据模型。对于多层嵌套的情况&#xff0c;比如二维数组&#xff0c;或者数…

pycharm远程连接docker容器

pycharm远程连接docker容器 1.根据镜像创建容器2.进入容器3.修改容器的root密码4. 容器安装openssh-server和openssh-client5.修改SSH配置文件6.重启ssh服务7. 退出测试8.配置pycharm并连接docker容器9. 选择docker环境 1.根据镜像创建容器 sudo docker run -itd --nameconn_t…

【目标检测】“复制-粘贴 copy-paste” 数据增强实现

文章目录 前言1. 效果展示代码说明3. 参考文档4. 不合适点 前言 本文来源论文《Simple Copy-Paste is a Strong Data Augmentation Method for Instance Segmentation》&#xff08;CVPR2020&#xff09;&#xff0c;对其数据增强方式进行实现。 论文地址&#xff1a;https:/…