如何保证消息不丢失
MQ的用途
- 异步发送(验证码,短信,邮件)
- MySQL,ES,Redis之间的数据同步
- 分布式事务
- 削峰填谷
消息可能丢失的环境
- 消息在产生端时候生产端挂掉,消息未到达交换机,消息丢失
- 消息在交换机未到达队列,消息丢失
- 消息队列中如果队列挂掉消息也可能丢失
- 消费者未接收消息,消息也可能丢失
针对消息丢失的解决方案:
生产者确认机制
RabbitMQ提供了 publisher confirm 机制来避免消息发送到MQ的过程中丢失,消息发送到MQ之后,会返回一个结果给发送者,表示消息发送是否处理成功。
如果发到交换机失败会返回一个nack publish-confirm
如果发送到消息队列失败会返回ack publish-return
消息发送失败的处理方式
- 回调方法即可重发
- 记录日志
- 保存到数据库然后定时发送,发送成功后即可删除表中的数据
消息持久化
MQ默认是内存存储消息,开启内存持久化功能可以确保缓存在MQ中的消息不丢失。
消息确认
RabbitMQ支持消费者确认机制,即:消费者处理消息后可以向MQ返回ack回执,MQ收到ack回执之后才会删除该消息
- 而SpringAMQP则允许配置三种确认机制:
- manual:手动ack,需要业务代码结束之后,调用api发送ack
- auto:自动ack,有Spring监测listener代码是否出现异常,没有响应则返回ack,抛出异常则返回nack
- none:关闭ack,MQ假定在获取消息后会成功处理,因此消息投递后立即被删除
我们可以利用Spring的retry机制,在消费者出现异常的时候利用本地重试,设置重试次数,当重试次数到了多少之后,如果消息依然失败,将消息投递带异常的交换机,交由人工处理。
RabbitMQ-如何保证消息不丢失?
- 开启生产者确认机制,确保生产者的消息能到达队列中
- 开启持久化功能,确保消息未消费钱在消息队列中不能丢失
- 开启消费确认机制未Auto,由Spring确认消息处理成功后完成ack
- 开启消费者失败重试机制,多次重试失败后将消息投递到异常交换机,交由人工处理
RabbitMQ消息重复消费问题
可能出现重复消费的原因:
- 网络抖动
- 消费者挂了
解决方案:(适用于任何MQ)
- 给每条消息设置唯一的标识id
- 幂等方案:【分布式锁,数据库锁(悲观锁,乐观锁)】
RabbitMQ中死信交换机
**问题:**RabbitMQ死信交换机?RabbitMQ延迟队列有了解过嘛?
- 延迟队列:进入队列的消息会被延迟消费的队列
- 场景:超时订单,限时优惠,定时发布
延迟队列=死信交换机+TTL(生存时间)
死信交换机
当一个队列中的消息满足下列情况之一时,可以被称为死信(dead letter)
- 消费者使用basic.reject或者basic.nack声明消费失败,并且将参数requence参数设置为false
- 消息是一个过期的消息,超时无人消费
- 要投递的队列消息堆积满了,最早的消息可能成为死信
如果该队列配置了dead-letter-exchange属性,执行了交换机,那么队列中的死信就会投递到这个交换机中,而这个交换机被称为死信交换机(Dead Letter Exchange,简称DLX)
TTL
TTL,也就是Time-To-Live。如果一个队列中的消息TTL结束仍未消费,则会变成死信,TTL超时分为两种情况:
- 消息所在的队列设置了存活时间
- 消息本身设置了存活时间
setExpiration(“5000”)表示消息存活的时间
延迟队列插件
DelayExchange插件,需要安装在RabbitMQ中
RabbitMQ有一个官方的插件社区,地址为:https://www.rabbitmq.com/community-plugins.html
延迟队列的插件
DelayExchange的本质还是官方的三种交换机,只是添加了延迟的功能,因此使用时需要声明一下交换机,交换机类型可以是任意的类型,然后设定delayed属性为true即可。
RabbitMQ中的死信交换机?(RabbitMQ延迟队列有了解过吗?)
- 我们当时一个业务使用到了延迟队列(超时订单,限时优惠,定时发布)
- 其中延迟队列就是用到了死信交换机和TTL(消息存活时间)实现的
- 消息超时未消费就变成了死信(死信的其他情况:拒绝消费,队列满了)
延迟队列插件实现延迟队列DelayExchange
- 声明一个交换机,添加delayed属性为true
- 发送消息时,添加x-delay头,值为超时时间
RabbitMQ消息堆积
RabbitMQ如果有100万消息堆积在MQ,如何解决(消息堆积怎么解决)
当生产者发送消息的速度超过了消费者处理消息的速度,就会导致队列中的消息堆积,指导队列存储消息达到上限,之后发送的消息就会成为死信,可能会被丢弃,这就是消息堆积问题。
解决的方式:
- 添加更多的消费者,提高消费速度
- 在消费者内开启线程池加快消息处理速度
- 扩大队列容积,提高堆积上限
惰性队列
- 接收消息后直接存储在磁盘中并非内存中
- 消费者要消费消息是才会从磁盘中读取到内存中
- 支持百万条消息存储
RabbitMQ如果有100万消息堆积在MQ,如何解决(消息堆积怎么解决)
解决消息队列的三种思路:
- 增加更多的消费者,提高消费的速度
- 在消费者内开启线程池加快消息处理的速度
- 扩大队列容积,提高堆积的上限,采用惰性队列
- 在声明队列的时候可以设置属性x-queue-mode为lazy,即为惰性队列
- 基于磁盘存储,消息上限高
- 性能比较稳定,但基于磁盘存储,受限与磁盘的IO,时效性降低
RabbitMQ高可用性
- 在生产环境下,使用集群来保证高可用性
- 普通集群,镜像集群,仲裁集群
普通集群
普通集群,或者交标准集群(classic cluster),具备以下特征:
-
会在集群的各个节点共享部分数据,包括:交换机,队列元信息。不包括队列中的消息。
-
当访问集群中的某个节点的时候,如果队列不在该节点,会从数据所在的节点传递到该节点并返回
-
队列所在的节点宕机,队列中的消息就会丢失
如果有一个消费者需要访问其中一个节点,而这个节点恰好宕机了,那么消息就会丢失
因此不能做到高可用。
镜像模式
镜像集群:本质是主从模式,具备以下的特征:
- 交换机,队列,队列中的消息会在各个MQ的镜像节点之间同步备份
- 创建队列的节点被称为该队列的主节点,备份到其他节点叫做该队列的镜像节点
- 一个队列的主机诶带你可能是另一个队列的镜像节点
- 所有的操作都是主节点完成,然后同步到镜像节点
- 主节点宕机后,镜像节点会替代新的主节点
仲裁队列
仲裁队列:是在3.8之后才有的新功能,用来替代镜像队列,具有下列的特征:
-
与镜像队列一样,都是主从模式,支持主从数据同步
-
使用非常简单,没有复杂的配置
-
主从同步基于Raft协议,强一致性
RabbitMQ的高可用性机制有了解过吗? -
生产环境下,我们主要采用的镜像模式搭建的集群,共有3个节点
-
镜像队列结构式一主多从(从就是就是镜像),所有的操作都是主节点完成的,完后同步给镜像节点
-
主节点宕机后,镜像节点会替代成新的主节点(如果在主节点同步完成前,主已经宕机了,可能会出现数据丢失的问题)
数据丢失的问题如何解决?
- 我们可以采用仲裁队列,与镜像队列一样,都是主从模式,支持主从数据同步,主从同步基于Raft协议,强一致性,并且使用起来也非常简单,不需要额外的配置,在声明队列的时候只要指定这个是仲裁队列即可
KafKa如何保证消息不丢失
Kafka如何保证消息不丢失
使用Kafka在消息收发的过程中都会出现消息丢失,Kafka分别给出了解决方案
- 生产者发送消息到Broker丢失
- 消息在Broker中存储丢失
- 消费者从Broker接收消息丢失
生产者发送消息到Broker丢失
- 设置异步发送(同步发送会导致线程阻塞)
消息发送失败的话记录日志 - 采用消息重试机制
消息在Broker中存储丢失
- 发送确认机制acks
确认机制 | 说明 |
---|---|
acks=0 | 生产者在成功写入消息之前不会等待任何来自服务器的响应,消息有丢失的风险,但是速度最快 |
acks=1(默认值) | 只要集群首领节点收到消息,生产者就会收到一个来自服务器的成功响应 |
acks=all | 只要当所有参与赋值的节点全部收到消息时,生产者才会收到一个来自服务器的成功响应 |
消费者从Broker接收消息丢失
- Kafka中的分区机制指的就是将每个主题划分成多个分区(Partition)
- Topic分区中消息只能由消费者组中的唯一一个消费者处理,不同的分区分配给不同的消费者(同一个消费者组)
不同的分区都是按照偏移量来存储数据的
并且消费者在消费之后会设置消费者自动提交偏移量,默认5秒一次,那么如果出现了重平衡的情况下,可能会造成重复消费和丢失数据问题。
如果消费者3负责p3队列的消费者,当消费者消费到6的时候消费者还没来得及提交偏移量,消费者挂了,那么系统会分配到别的消费者根据偏移量进行消费,那么就会出现问题,分配过来的消费者1就会从4开始消费,那么就会出现重复消费的问题。
解决方案:
-
禁止自动提交偏移量,改为手动提交
-
同步提交
-
异步提交
-
同步+异步组合提交
Kafka如何保证消息不丢失?
需要从以下的三个方面去解决这个问题: -
生产者发送消息到Broker丢失
- 设置异步发送,发送失败使用回调进行记录和重发
- 失败重试,参数配置,可以设置重试次数
-
消息在Broker中存储丢失
- 发送acks,选择all,让所有的副本都保存数据之后确认
-
消费者从Broker中接收消息丢失
- 关闭自动提交偏移量,开启手动提交偏移量
- 提交方式,最好是同步+异步提交
Kafka中消息的重复消费问题如何解决?
-
关闭自动提交偏移量,开启手动提交偏移量
-
提交方式,最好是同步+异步
-
幂等方案
Kafka如何保证消息顺序性
应用场景:
- 即时消息中的单对单聊天和群聊,保证消息发送方消息发送和接受方的顺序一致性
- 充值转账两个渠道在同一时间进行余额变更,短信通知必须要顺序
Topic分区中消息只能由消费者组中的唯一一个消费者处理,所以消息肯定是按照先后顺序进行处理的,但是它也仅仅是保证Topic的一个分区顺序处理,不能保证跨分区的消息先后处理顺序,所以如果你想要顺序处理Topic的所有消息,那就只提供一个分区。
Kafka如何保证消费的顺序性?
问题原因:
一个Topic的数据可能存储在不同的分区中,每个分区都有一个按照顺序的存储的偏移量,如果消费者关联了多个分区不能保证顺序性。
解决方案: - 发送消息时制定分区号
- 发送消息时按照相同的业务设置相同的key
Kafka高可用机制
Kafka的高可用机制了解吗?
- Kafka的服务端被称为Broker的服务进行构成,即一个Kafka集群由多个Broker组成
- 这样如果集群中某一台机器宕机,其他的机器上的Broker也依然能够对外提供服务,这其实就是Kafka提供的高可用的手段之一
分区备份机制:
假设某一个topic中有三个分区p0,p1,p2 - 一个Topic中有多个分区,每个分区有多个副本,其中一个是leader,其他的都是flower,副本存储在不同的Broker中
- 所有的分区副本的内容都是相同的,如果发生leader发生故障的时候,会自动将其中的一个follower提升为leader
分区备份机制
ISR(in-sync replica)需要同步复制保存的follower
如果leader失效后,需要选取新的leader,选举原则如下:
第一:选举是有限从ISR中选定,因为这个列表中follower的数据时与leader同步的
第二:如果ISR列表中follower都不行,就只能从其他follower中选取
Kafka的高可用性机制了解过吗?
可以从两个层面回答,第一是集群,第二个是复制
集群:
一个Kafka集群由多个Broker实例组成,即时某一台宕机,也不能耽误其他Broker继续对外提供服务
复制机制:
- 一个Topic由多个分区,每个分区有多个副本,有一个leader,其余都是follower,副本存储在不同的Broker
- 所有的分区副本的内容都是相同的,如果leader发生故障时,会自动将其中follower提升为leader,保证系统的容错性,高可用性。
解释一下复制机制中的ISR
ISR(in-sync replica)需要同步复制保存的follower
分区副本分为两类,一个ISR,与leader副本同步保存数据,另外一个普通副本,是异步同步数据,当leader挂掉之后,会优先从ISR副本列表中选取一个作为leader。
Kafka数据清理机制
- Kafka文件存储机制
- Kafka数据清理机制
Kafka文件存储机制
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-svasRlbJ-1687243093845)(assets/image-20230523151425403.png)]
Kafka采用了分段存储的方式存储数据
为什么采用分段呢?
- 删除无用的文件更加方便,提高磁盘利用率
- 查找数据更便捷
默认7天清理一次
数据清理机制
日志清理策略有两个
-
根据消息的保留时间,当消息在Kafka中保存的时间超过了指定的时间,就会触发清理程序,默认7天
-
根据topic存储的数据大小,当topic所占的日志文件大小大于一定的阈值,则开始删除最久的消息,需手动开启
Kafka数据清理机制了解过吗?
Kafka存储结构
- Kafka中topic的数据存储在分区上,分区如果文件过大会分段存储segment
- 每个分段都在磁盘上以索引(xxxx.index)和日志文件(xxxx.log)的形式存储
- 分段的好处是,第一能够尖山单个文件内容的大小,查找数据方便,第二方便Kafka进行日志清理
日志清理的策略有两个:
- 根据消息的保留时间,当消息保存时间超过了指定的时间,就会触发清理,默认是168小时(7天)
- 根据topic存储数据大小,当topic所占的日志文件大小大于一定的阈值,则开始删除最久的消息。(默认关闭)
Kafka高性能设计
Kafka的高性能设计了解过吗?
- 消息分区:不受单台服务器的限制,可以不受限制的处理更多的数据
- 顺序读写:磁盘顺序读写,提高读写性能
- 页缓存:把磁盘中的数据缓存到内存中,把从磁盘的访问变为对内存的访问
- 零拷贝:减少上下文切换及数据拷贝
- 消息压缩:减少磁盘IO和网络IO
- 分批发送:将消息打包批量发送,减少网络开销
零拷贝
Linux中传统Kafka的数据消费的方式(一共要经过4次拷贝性能并不高)
零拷贝技术:
当确定消费者之后直接将任务委托给Linux系统进行操作,这样只拷贝了两次
Kafka中实现高性能的设计了解过吗?
- 消息分区:不受单台服务器的限制,可以不受限制的处理更多的数据
- 顺序读写:磁盘顺序读写,提高读写的效率
- 页缓存:把磁盘中的数据缓存到内存中,把对磁盘的访问变为对内存的访问
- 零拷贝:减少磁盘IO和网络IO
- 分批发送:将消息打包批量发送,减少网络开销
笔记是对黑马课程中的知识进行的个人总结,图片借鉴了课程视频中的资料,感谢黑马程序员的开源精神,哈哈,如有问题联系我删除!