面试题:Kafka 为什么会丢消息?
文章目录
- 1、如何知道有消息丢失?
- 2、哪些环节可能丢消息?
- 3、如何确保消息不丢失?
引入 MQ 消息中间件最直接的目的:系统解耦以及流量控制(削峰填谷)
- 系统解耦: 上下游系统之间的通信相互依赖,利用 MQ 消息队列可以隔离上下游环境变化带来的不稳定因素。
- 流量控制: 超高并发场景中,引入 MQ 可以实现流量 “削峰填谷” 的作用以及服务异步处理,不至于打崩服务。
引入 MQ 同样带来其他问题:数据一致性。
在分布式系统中,如果两个节点之间存在数据同步,就会带来数据一致性的问题。消息生产端发送消息到 MQ 再到消息消费端需要保证消息不丢失。
所以在使用 MQ 消息队列时,需要考虑这 3 个问题:
- 如何知道有消息丢失?
- 哪些环节可能丢消息?
- 如何确保消息不丢失?
1、如何知道有消息丢失?
如何感知消息是否丢失了?可总结如下:
- 他人反馈: 运营、PM 反馈消息丢失。
- 监控报警: 监控指定指标,即时报警人工调整。Kafka 集群异常、Broker 宕机、Broker 磁盘挂载问题、消费者异常导致消息积压等都会给用户直接感觉是消息丢失了。
案例:舆情分析中数据采集同步
- PM 可自己下发采集调度指令,去采集特定数据。
- PM 可通过 ES 近实时查询对应数据,若没相应数据可再次下发指令。
当感知消息丢失了,那就需要一种机制来检查消息是否丢失。
检索消息
运维工具有:
1.查看 Kafka 消费位置:
# 查看某个topic的message数量
$ ./kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic test_topic# 查看consumer Group列表,面试宝典:https://www.yoodb.com
$ ./kafka-consumer-groups.sh --list --bootstrap-server 192.168.88.108:9092# 查看 offset 消费情况
$ ./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group console-consumer-1152 --describe
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
console-consumer-1152 test_topic 0 - 4 - consumer-console-consumer-1152-1-2703ea2b-b62d-4cfd-8950-34e8c321b942 /127.0.0.1 consumer-console-consumer-1152-1
2.利用工具:Kafka Tools
3.其他可见化界面工具
2、哪些环节可能丢消息?
一条消息从生产到消费完成经历 3 个环节:消息生产者、消息中间件、消息消费者。
哪个环节都有可能出现消息丢失问题。
1)生产端
首先要认识到 Kafka 生产端发送消息流程:
调用 send() 方法时,不会立刻把消息发送出去,而是缓存起来,选择恰当时机把缓存里的消息划分成一批数据,通过 Sender 线程按批次发送给服务端 Broker。
此环节丢失消息的场景有: 即导致 Producer 消息没有发送成功
1.网络波动: 生产者与服务端之间的链路不可达,发送超时。现象是:各端状态正常,但消费端就是没有消费消息,就像丢失消息一样。
2.解决措施: 重试 props.put(“retries”, “10”);
3.不恰当配置: 发送消息无 ack 确认; 发送消息失败无回调,无日志。
producer.send(new ProducerRecord<>(topic, messageKey, messageStr), new CallBack(){...});
4.解决措施: *设置 acks=1 或者 acks=all。发送消息设置回调。
回顾下重要的参数: acks
- acks=0:不需要等待服务器的确认. 这是 retries设置无效. 响应里来自服务端的 offset 总是 -1,producer只管发不管发送成功与否。延迟低,容易丢失数据。
- acks=1:表示 leader 写入成功(但是并没有刷新到磁盘)后即向 producer 响应。延迟中等,一旦 leader 副本挂了,就会丢失数据。
- acks=all:等待数据完成副本的复制, 等同于 -1. 假如需要保证消息不丢失, 需要使用该设置. 同时需要设置 unclean.leader.election.enable 为 true, 保证当 ISR 列表为空时, 选择其他存活的副本作为新的 leader.
2)服务端
先来了解下 Kafka Broker 写入数据的过程:
1.Broker接收到一批数据,会先写入内存 PageCache(OS Cache)中。
2.操作系统会隔段时间把 OS Cache 中数据进行刷盘,这个过程会是 「异步批量刷盘」 。
这里就有个隐患,如果数据写入 PageCache 后 Kafka Broker宕机会怎样?机子宕机/掉电?
Kafka Broker 宕机: 消息不会丢失。因为数据已经写入 PageCache,只等待操作系统刷盘即可。
机子宕机/掉电: 消息会丢失。因为数据仍在内存里,内存RAM掉电后就会丢失数据。
- 解决方案 :使用带蓄电池后备电源的缓存 cache,防止系统断电异常。
1.对比学习 MySQL 的 “双1” 策略,基本不使用这个策略,因为 “双1” 会导致频繁的 I/O 操作,也是最慢的一种。
2.对比学习 Redis 的 AOF 策略,默认且推荐的策略:Everysec(AOF_FSYNC_EVERYSEC) 每一秒钟保存一次(默认): 。每个写命令执行完,
只是先把日志写到 AOF 文件的内存缓冲区, 每隔一秒把缓冲区中的内容写入磁盘。
拓展:Kafka 日志刷盘机制
# 推荐采用默认值,即不配置该配置,交由操作系统自行决定何时落盘,以提升性能。
# 针对 broker 配置:
log.flush.interval.messages=10000 # 日志落盘消息条数间隔,即每接收到一定条数消息,即进行log落盘。
log.flush.interval.ms=1000 # 日志落盘时间间隔,单位ms,即每隔一定时间,即进行log落盘。# 针对 topic 配置:面试宝典:https://www.yoodb.com
flush.messages.flush.ms=1000 # topic下每1s刷盘
flush.messages=1 # topic下每个消息都落盘# 查看 Linux 后台线程执行配置
$ sysctl -a | grep dirty
vm.dirty_background_bytes = 0
vm.dirty_background_ratio = 10 # 表示当脏页占总内存的的百分比超过这个值时,后台线程开始刷新脏页。
vm.dirty_bytes = 0
vm.dirty_expire_centisecs = 3000 # 表示脏数据多久会被刷新到磁盘上(30秒)。
vm.dirty_ratio = 20
vm.dirty_writeback_centisecs = 500 # 表示多久唤醒一次刷新脏页的后台线程(5秒)。
vm.dirtytime_expire_seconds = 43200
Broker 的可靠性需要依赖其多副本机制: 一般副本数 3 个(配置参数:replication.factor=3)
- Leader Partition 副本:提供对外读写机制。
- Follower Partition 副本:同步 Leader 数据。
副本之间的数据同步也可能出现问题:数据丢失问题和数据不一致问题。
解决方案:ISR 和 Epoch 机制
-
ISR(In-Sync Replicas) : 当 Le``ader 宕机,可以从 ISR 中选择一个 Follower 作为 Leader。
-
Epoch 机制: 解决 Leader 副本高水位更新和 Follower 副本高水位更新在时间上是存在错配问题。
Tips: Kafka 0.11.x 版本才引入 leader epoch 机制解决高水位机制弊端。
对应需要的配置参数如下:
1.acks=-1 或者 acks=all: 必须所有副本均同步到消息,才能表明消息发送成功。
2.replication.factor >= 3: 副本数至少有 3 个。
3.min.insync.replicas > 1: 代表消息至少写入 2个副本才算发送成功。前提需要 acks=-1。
举个栗子:Leader 宕机了,至少要保证 ISR 中有一个 Follower,这样这个Follwer被选举为Leader 且不会丢失数据。
公式:replication.factor = min.insync.replicas + 1
4.unclean.leader.election.enable=false: 防止不在 ISR 中的 Follower 被选举为 Leader。
Kafka 0.11.0.0版本开始默认 unclean.leader.election.enable=false
3)消费端
消费端消息丢失场景有:
1.消息堆积: 几个分区的消息都没消费,就跟丢消息一样。
2.解决措施: 一般问题都出在消费端,尽量提高客户端的消费速度,消费逻辑另起线程进行处理。
3.自动提交: 消费端拉下一批数据,正在处理中自动提交了 offset,这时候消费端宕机了; 重启后,拉到新一批数据,而上一批数据却没处理完。
4.解决措施: 取消自动提交 auto.commit = false,改为手动 ack。
5.心跳超时,引发 Rebalance: 客户端心跳超时,触发 Rebalance被踢出消费组。如果只有这一个客户端,那消息就不会被消费了。
同时避免两次 poll 的间隔时间超过阈值:
6.max.poll.records:降低该参数值,建议远远小于 <单个线程每秒消费的条数> * <消费线程的个数> * <max.poll.interval.ms> 的积。
max.poll.interval.ms: 该值要大于 <max.poll.records> / (<单个线程每秒消费的条数> * <消费线程的个数>) 的值。
解决措施: 客户端版本升级至 0.10.2 以上版本。
案例:凡凡曾遇到数据同步时,消息中的文本需经过 NLP 的 NER 分析,再同步到 ES。
这个过程的主要流程是:
1.数据同步程序从 Kafka 中拉取消息。
2.数据同步程序将消息内的文本发送的 NER 进行分析,得到特征数组。
3.数据同步程序将消息同步给 ES。
现象:线上数据同步程序运行一段时间后,消息就不消费了。
- 排查日志: 发现有 Rebalance 日志,怀疑是客户端消费太慢被踢出了消费组。
- 本地测试: 发现运行一段时间也会出现 Rebalance,且 NLP的NER 服务访问 HTTP 500 报错。
- 得出结论: 因NER服务异常,导致数据同步程序消费超时。且当时客户端版本为 v0.10.1,Consumer 没有独立线程维持心跳,而是把心跳维持与 poll 接口耦合在一起,从而也会造成心跳超时。
当时解决措施是:
1.session.timeout.ms: 设置为 25s,当时没有升级客户端版本,怕带来其他问题。
2.熔断机制: 增加 Hystrix,超过 3 次服务调用异常就熔断,保护客户端正常消费数据。
3、如何确保消息不丢失?
掌握这些技能:
1.熟悉消息从发送到消费的每个阶段
2.监控报警 Kafka 集群
3.熟悉方案 “MQ 可靠消息投递”
怎么确保消息 100% 不丢失?
到这,总结下:
①生产端:
- 设置重试:props.put(“retries”, “10”);
- 设置 acks=all
- 设置回调:producer.send(msg, new CallBack(){…});
②Broker:
- 内存:使用带蓄电池后备电源的缓存 cache。
- Kafka 版本 0.11.x 以上:支持 Epoch 机制。
- replication.factor >= 3: 副本数至少有 3 个。
- min.insync.replicas > 1: 代表消息至少写入 2个副本才算发送成功。前提需要 acks=-1。
- unclean.leader.election.enable=false: 防止不在 ISR 中的 Follower 被选举为 Leader。
③消费端
- 客户端版本升级至 0.10.2 以上版本。
- 取消自动提交 auto.commit = false,改为手动 ack。
- 尽量提高客户端的消费速度,消费逻辑另起线程进行处理
相关文章:

面试题:Kafka 为什么会丢消息?
文章目录 1、如何知道有消息丢失?2、哪些环节可能丢消息?3、如何确保消息不丢失? 引入 MQ 消息中间件最直接的目的:系统解耦以及流量控制(削峰填谷) 系统解耦: 上下游系统之间的通信相互依赖&a…...

WSL安装异常:WslRegisterDistribution failed with error: 0xc03a001a
简介:如果文件夹右上角是否都有两个相对的蓝色箭头,在进行安装wsl时,设置就会抛出 Installing WslRegisterDistribution failed with error: 0xc03a001a的异常 历史攻略: 卸载WSL WSL:运行Linux文件 WSL࿱…...
【C语言 模拟实现strcmp函数】
C语言程序设计笔记---025 C语言之模拟实现strcmp函数1、介绍strcmp函数2、模拟实现strcmp函数3、结语 C语言之模拟实现strcmp函数 前言: 通过C语言字符串函数的知识,这篇将对strcmp函数进行深入学习底层原理的知识,并模拟实现对应功能。 /知…...
maven 依赖版本冲突异常
maven 依赖版本冲突异常 好巧不巧,前几天刚刚复习完 maven 的内容今天就碰到 maven 报错。 起因是这样的,项目马上快要上线了,在上线之前需要跑一些 audit 去检查项目是否安全(这里主要是 outdated 的依赖检查)。总体…...
蓝牙核心规范(V5.4)11.5-LE Audio 笔记之Context Type
专栏汇总网址:蓝牙篇之蓝牙核心规范学习笔记(V5.4)汇总_蓝牙核心规范中文版_心跳包的博客-CSDN博客 爬虫网站无德,任何非CSDN看到的这篇文章都是盗版网站,你也看不全。认准原始网址。!!! 蓝牙中的上下文类型(Context Type)是用于描述音频流当前使用情况或相关使用情…...

【Linux】RPM包使用详解
🍁 博主 "开着拖拉机回家"带您 Go to New World.✨🍁 🦄 个人主页——🎐开着拖拉机回家_大数据运维-CSDN博客 🎐✨🍁 🪁🍁 希望本文能够给您带来一定的帮助🌸文…...
勒索病毒最新变种.Elbie勒索病毒来袭,如何恢复受感染的数据?
引言: 网络犯罪正变得越来越隐秘和危险。其中,.Elbie勒索病毒作为数字犯罪的一部分,以其阴险和复杂性而备受关注。本文将带您深入探索.Elbie勒索病毒的工作原理和如何应对这一数字迷宫。如果受感染的数据确实有恢复的价值与必要性࿰…...
ArduPilot开源飞控之AP_Mission
ArduPilot开源飞控之AP_Mission 1. 源由2. AP_Mission类3 简令结构3.1 导航相关3.1.1 jump command3.1.2 condition delay command3.1.3 condition distance command3.1.4 condition yaw command3.1.5 change speed command3.1.6 nav guided command3.1.7 do VTOL transition3.…...

JVM111
JVM1 字节码与多语言混合编程 字节码 我们平时说的java字节码, 指的是用java语言编译成的字节码。准确的说任何能在jvm平台上执行的字节码格式都是一样的。所以应该统称为:jvm字节码。不同的编译器,可以编译出相同的字节码文件,字节码文件…...

排序篇(三)----交换排序
排序篇(三)----交换排序 1.冒泡排序 基本思想: 通过不断地比较相邻的元素,将较大的元素往后移动,从而实现排序的目的。 具体的步骤如下: 从待排序的数组中选择相邻的两个元素进行比较,如果前一个元素大于后一个元素&#…...

React antd Table点击下一页后selectedRows丢失之前页选择内容的问题
一、问题 使用了React antd 的<Table>标签,是这样记录选中的行id与行内容的: <TabledataSource{data.list}rowSelection{{selectedRowKeys: selectedIdsInSearchTab,onChange: this.onSelectChange,}} // 表格是否可复选,加 type: …...
蓝牙核心规范(V5.4)11.4-LE Audio 笔记之音频模型
专栏汇总网址:蓝牙篇之蓝牙核心规范学习笔记(V5.4)汇总_蓝牙核心规范中文版_心跳包的博客-CSDN博客 爬虫网站无德,任何非CSDN看到的这篇文章都是盗版网站,你也看不全。认准原始网址。!!! 从一开始,蓝牙低功耗(Bluetooth Low Energy,BLE)音频的开发就秉持着“以设…...

Spring Boot:利用JPA进行数据库的查删
目录标题 DAO 、Service 、 Controller 层控制器文件示例代码-单个查找查找成功示例代码-列表查找查找成功示例代码-删除删除成功 DAO 、Service 、 Controller 层 DAO 层负责数据库访问,它封装了对数据库的访问操作,例如查询、插入、更新和删除等。 Q…...
1711: 【穷举】满足条件的整数
题目描述 假设a、b、c均为整数(1<a,b,c<100),同时a<b,找出所有符合条件:a2 b2 n*c3的整数组。 按a从小到大的顺序输出所有满足条件的整数组(若a相同,则按b从小到大的顺序输出) 输入…...

【数据结构】堆的应用-----TopK问题
目录 一、前言 二、Top-k问题 💦解法一:暴力排序 💦解法二:建立N个数的堆 💦解法三:建立K个数的堆(最优解) 三、完整代码和视图 四、共勉 一、前言 在之前的文章中ÿ…...
QT之xml文件的读写
QT之xml文件的读写 简介用法举例 简介 QT的QDomDocument、QDomElement、QDomNode是Qt XML模块中的三个类,用于解析和操作XML文档。 1)QDomDocument类: QDomDocument类表示整个XML文档。它提供了解析XML文档的方法,如setContent(…...
C语言中的异常处理机制是什么?
C语言中的异常处理机制 C语言是一门强大而灵活的编程语言,它为程序员提供了广泛的控制权和自由度。然而,C语言本身并不提供像其他高级语言一样的内置异常处理机制,如Java中的try-catch或Python中的异常处理。因此,C语言程序员需要…...
Java中的并发编程模型和常用工具类
本文主要介绍了Java中的并发编程模型和常用工具类,首先阐述了并发编程的概念及其重要性,然后详细介绍了线程的基本概念、生命周期和状态转换、同步与互斥、死锁问题以及线程池的使用和实现原理。接着介绍了synchronized关键字和Lock接口的使用、原子变量…...

第10章 MySQL(一)
10.1 谈谈MySQL的架构 难度:★★ 重点:★ 白话解析 要想彻底的理解MySQL,它的架构一定要先弄清楚,当Java程序员通过JDBC或者Mybatis去执行一条SQL的时候,到底经历了什么。下边先看一幅图: 户端:Java程序员通过JDBC或者Mybatis去拿MySQL的驱动程序,实际上就是拿客户端。…...

英飞凌 Tricore 架构中断系统详解
本文以TC3系列MCU为例,先来了解中断源是如何产生的,再看一下CPU是如何处理中断源的。 AURIX TC3XX的中断路由模块 Interrupt Router (IR) 在TC3中,中断既可以被CPU处理,也可以被DMA处理,所以手册中不再把中断称为中断…...

docker详细操作--未完待续
docker介绍 docker官网: Docker:加速容器应用程序开发 harbor官网:Harbor - Harbor 中文 使用docker加速器: Docker镜像极速下载服务 - 毫秒镜像 是什么 Docker 是一种开源的容器化平台,用于将应用程序及其依赖项(如库、运行时环…...

css实现圆环展示百分比,根据值动态展示所占比例
代码如下 <view class""><view class"circle-chart"><view v-if"!!num" class"pie-item" :style"{background: conic-gradient(var(--one-color) 0%,#E9E6F1 ${num}%),}"></view><view v-else …...
rknn优化教程(二)
文章目录 1. 前述2. 三方库的封装2.1 xrepo中的库2.2 xrepo之外的库2.2.1 opencv2.2.2 rknnrt2.2.3 spdlog 3. rknn_engine库 1. 前述 OK,开始写第二篇的内容了。这篇博客主要能写一下: 如何给一些三方库按照xmake方式进行封装,供调用如何按…...
k8s从入门到放弃之Ingress七层负载
k8s从入门到放弃之Ingress七层负载 在Kubernetes(简称K8s)中,Ingress是一个API对象,它允许你定义如何从集群外部访问集群内部的服务。Ingress可以提供负载均衡、SSL终结和基于名称的虚拟主机等功能。通过Ingress,你可…...

遍历 Map 类型集合的方法汇总
1 方法一 先用方法 keySet() 获取集合中的所有键。再通过 gey(key) 方法用对应键获取值 import java.util.HashMap; import java.util.Set;public class Test {public static void main(String[] args) {HashMap hashMap new HashMap();hashMap.put("语文",99);has…...
mongodb源码分析session执行handleRequest命令find过程
mongo/transport/service_state_machine.cpp已经分析startSession创建ASIOSession过程,并且验证connection是否超过限制ASIOSession和connection是循环接受客户端命令,把数据流转换成Message,状态转变流程是:State::Created 》 St…...
Java - Mysql数据类型对应
Mysql数据类型java数据类型备注整型INT/INTEGERint / java.lang.Integer–BIGINTlong/java.lang.Long–––浮点型FLOATfloat/java.lang.FloatDOUBLEdouble/java.lang.Double–DECIMAL/NUMERICjava.math.BigDecimal字符串型CHARjava.lang.String固定长度字符串VARCHARjava.lang…...
Frozen-Flask :将 Flask 应用“冻结”为静态文件
Frozen-Flask 是一个用于将 Flask 应用“冻结”为静态文件的 Python 扩展。它的核心用途是:将一个 Flask Web 应用生成成纯静态 HTML 文件,从而可以部署到静态网站托管服务上,如 GitHub Pages、Netlify 或任何支持静态文件的网站服务器。 &am…...

如何将联系人从 iPhone 转移到 Android
从 iPhone 换到 Android 手机时,你可能需要保留重要的数据,例如通讯录。好在,将通讯录从 iPhone 转移到 Android 手机非常简单,你可以从本文中学习 6 种可靠的方法,确保随时保持连接,不错过任何信息。 第 1…...
JVM暂停(Stop-The-World,STW)的原因分类及对应排查方案
JVM暂停(Stop-The-World,STW)的完整原因分类及对应排查方案,结合JVM运行机制和常见故障场景整理而成: 一、GC相关暂停 1. 安全点(Safepoint)阻塞 现象:JVM暂停但无GC日志,日志显示No GCs detected。原因:JVM等待所有线程进入安全点(如…...