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

面试题:Kafka 为什么会丢消息?

文章目录

  • 1、如何知道有消息丢失?
  • 2、哪些环节可能丢消息?
  • 3、如何确保消息不丢失?


引入 MQ 消息中间件最直接的目的:系统解耦以及流量控制(削峰填谷)

  • 系统解耦: 上下游系统之间的通信相互依赖,利用 MQ 消息队列可以隔离上下游环境变化带来的不稳定因素。
  • 流量控制: 超高并发场景中,引入 MQ 可以实现流量 “削峰填谷” 的作用以及服务异步处理,不至于打崩服务。

引入 MQ 同样带来其他问题:数据一致性。

在分布式系统中,如果两个节点之间存在数据同步,就会带来数据一致性的问题。消息生产端发送消息到 MQ 再到消息消费端需要保证消息不丢失。

在这里插入图片描述
所以在使用 MQ 消息队列时,需要考虑这 3 个问题:

  • 如何知道有消息丢失?
  • 哪些环节可能丢消息?
  • 如何确保消息不丢失?

在这里插入图片描述

1、如何知道有消息丢失?

如何感知消息是否丢失了?可总结如下:

  1. 他人反馈: 运营、PM 反馈消息丢失。
  2. 监控报警: 监控指定指标,即时报警人工调整。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 总是 -1producer只管发不管发送成功与否。延迟低,容易丢失数据。
  • acks=1:表示 leader 写入成功(但是并没有刷新到磁盘)后即向 producer 响应。延迟中等,一旦 leader 副本挂了,就会丢失数据。
  • acks=all:等待数据完成副本的复制, 等同于 -1. 假如需要保证消息不丢失, 需要使用该设置. 同时需要设置 unclean.leader.election.enabletrue, 保证当 ISR 列表为空时, 选择其他存活的副本作为新的 leader.

2)服务端
先来了解下 Kafka Broker 写入数据的过程:

1.Broker接收到一批数据,会先写入内存 PageCacheOS 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、如何知道有消息丢失&#xff1f;2、哪些环节可能丢消息&#xff1f;3、如何确保消息不丢失&#xff1f; 引入 MQ 消息中间件最直接的目的&#xff1a;系统解耦以及流量控制&#xff08;削峰填谷&#xff09; 系统解耦&#xff1a; 上下游系统之间的通信相互依赖&a…...

WSL安装异常:WslRegisterDistribution failed with error: 0xc03a001a

简介&#xff1a;如果文件夹右上角是否都有两个相对的蓝色箭头&#xff0c;在进行安装wsl时&#xff0c;设置就会抛出 Installing WslRegisterDistribution failed with error: 0xc03a001a的异常 历史攻略&#xff1a; 卸载WSL WSL&#xff1a;运行Linux文件 WSL&#xff1…...

【C语言 模拟实现strcmp函数】

C语言程序设计笔记---025 C语言之模拟实现strcmp函数1、介绍strcmp函数2、模拟实现strcmp函数3、结语 C语言之模拟实现strcmp函数 前言&#xff1a; 通过C语言字符串函数的知识&#xff0c;这篇将对strcmp函数进行深入学习底层原理的知识&#xff0c;并模拟实现对应功能。 /知…...

maven 依赖版本冲突异常

maven 依赖版本冲突异常 好巧不巧&#xff0c;前几天刚刚复习完 maven 的内容今天就碰到 maven 报错。 起因是这样的&#xff0c;项目马上快要上线了&#xff0c;在上线之前需要跑一些 audit 去检查项目是否安全&#xff08;这里主要是 outdated 的依赖检查&#xff09;。总体…...

蓝牙核心规范(V5.4)11.5-LE Audio 笔记之Context Type

专栏汇总网址:蓝牙篇之蓝牙核心规范学习笔记(V5.4)汇总_蓝牙核心规范中文版_心跳包的博客-CSDN博客 爬虫网站无德,任何非CSDN看到的这篇文章都是盗版网站,你也看不全。认准原始网址。!!! 蓝牙中的上下文类型(Context Type)是用于描述音频流当前使用情况或相关使用情…...

【Linux】RPM包使用详解

&#x1f341; 博主 "开着拖拉机回家"带您 Go to New World.✨&#x1f341; &#x1f984; 个人主页——&#x1f390;开着拖拉机回家_大数据运维-CSDN博客 &#x1f390;✨&#x1f341; &#x1fa81;&#x1f341; 希望本文能够给您带来一定的帮助&#x1f338;文…...

勒索病毒最新变种.Elbie勒索病毒来袭,如何恢复受感染的数据?

引言&#xff1a; 网络犯罪正变得越来越隐秘和危险。其中&#xff0c;.Elbie勒索病毒作为数字犯罪的一部分&#xff0c;以其阴险和复杂性而备受关注。本文将带您深入探索.Elbie勒索病毒的工作原理和如何应对这一数字迷宫。如果受感染的数据确实有恢复的价值与必要性&#xff0…...

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字节码&#xff0c; 指的是用java语言编译成的字节码。准确的说任何能在jvm平台上执行的字节码格式都是一样的。所以应该统称为:jvm字节码。不同的编译器&#xff0c;可以编译出相同的字节码文件&#xff0c;字节码文件…...

排序篇(三)----交换排序

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

React antd Table点击下一页后selectedRows丢失之前页选择内容的问题

一、问题 使用了React antd 的<Table>标签&#xff0c;是这样记录选中的行id与行内容的&#xff1a; <TabledataSource{data.list}rowSelection{{selectedRowKeys: selectedIdsInSearchTab,onChange: this.onSelectChange,}} // 表格是否可复选&#xff0c;加 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 层负责数据库访问&#xff0c;它封装了对数据库的访问操作&#xff0c;例如查询、插入、更新和删除等。 Q…...

1711: 【穷举】满足条件的整数

题目描述 假设a、b、c均为整数&#xff08;1<a,b,c<100)&#xff0c;同时a<b&#xff0c;找出所有符合条件&#xff1a;a2 b2 n*c3的整数组。 按a从小到大的顺序输出所有满足条件的整数组&#xff08;若a相同&#xff0c;则按b从小到大的顺序输出&#xff09; 输入…...

【数据结构】堆的应用-----TopK问题

目录 一、前言 二、Top-k问题 &#x1f4a6;解法一&#xff1a;暴力排序 &#x1f4a6;解法二&#xff1a;建立N个数的堆 &#x1f4a6;解法三&#xff1a;建立K个数的堆&#xff08;最优解&#xff09; 三、完整代码和视图 四、共勉 一、前言 在之前的文章中&#xff…...

QT之xml文件的读写

QT之xml文件的读写 简介用法举例 简介 QT的QDomDocument、QDomElement、QDomNode是Qt XML模块中的三个类&#xff0c;用于解析和操作XML文档。 1&#xff09;QDomDocument类&#xff1a; QDomDocument类表示整个XML文档。它提供了解析XML文档的方法&#xff0c;如setContent(…...

C语言中的异常处理机制是什么?

C语言中的异常处理机制 C语言是一门强大而灵活的编程语言&#xff0c;它为程序员提供了广泛的控制权和自由度。然而&#xff0c;C语言本身并不提供像其他高级语言一样的内置异常处理机制&#xff0c;如Java中的try-catch或Python中的异常处理。因此&#xff0c;C语言程序员需要…...

Java中的并发编程模型和常用工具类

本文主要介绍了Java中的并发编程模型和常用工具类&#xff0c;首先阐述了并发编程的概念及其重要性&#xff0c;然后详细介绍了线程的基本概念、生命周期和状态转换、同步与互斥、死锁问题以及线程池的使用和实现原理。接着介绍了synchronized关键字和Lock接口的使用、原子变量…...

第10章 MySQL(一)

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

英飞凌 Tricore 架构中断系统详解

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

Chapter03-Authentication vulnerabilities

文章目录 1. 身份验证简介1.1 What is authentication1.2 difference between authentication and authorization1.3 身份验证机制失效的原因1.4 身份验证机制失效的影响 2. 基于登录功能的漏洞2.1 密码爆破2.2 用户名枚举2.3 有缺陷的暴力破解防护2.3.1 如果用户登录尝试失败次…...

阿里云ACP云计算备考笔记 (5)——弹性伸缩

目录 第一章 概述 第二章 弹性伸缩简介 1、弹性伸缩 2、垂直伸缩 3、优势 4、应用场景 ① 无规律的业务量波动 ② 有规律的业务量波动 ③ 无明显业务量波动 ④ 混合型业务 ⑤ 消息通知 ⑥ 生命周期挂钩 ⑦ 自定义方式 ⑧ 滚的升级 5、使用限制 第三章 主要定义 …...

关于nvm与node.js

1 安装nvm 安装过程中手动修改 nvm的安装路径&#xff0c; 以及修改 通过nvm安装node后正在使用的node的存放目录【这句话可能难以理解&#xff0c;但接着往下看你就了然了】 2 修改nvm中settings.txt文件配置 nvm安装成功后&#xff0c;通常在该文件中会出现以下配置&…...

拉力测试cuda pytorch 把 4070显卡拉满

import torch import timedef stress_test_gpu(matrix_size16384, duration300):"""对GPU进行压力测试&#xff0c;通过持续的矩阵乘法来最大化GPU利用率参数:matrix_size: 矩阵维度大小&#xff0c;增大可提高计算复杂度duration: 测试持续时间&#xff08;秒&…...

企业如何增强终端安全?

在数字化转型加速的今天&#xff0c;企业的业务运行越来越依赖于终端设备。从员工的笔记本电脑、智能手机&#xff0c;到工厂里的物联网设备、智能传感器&#xff0c;这些终端构成了企业与外部世界连接的 “神经末梢”。然而&#xff0c;随着远程办公的常态化和设备接入的爆炸式…...

laravel8+vue3.0+element-plus搭建方法

创建 laravel8 项目 composer create-project --prefer-dist laravel/laravel laravel8 8.* 安装 laravel/ui composer require laravel/ui 修改 package.json 文件 "devDependencies": {"vue/compiler-sfc": "^3.0.7","axios": …...

Linux离线(zip方式)安装docker

目录 基础信息操作系统信息docker信息 安装实例安装步骤示例 遇到的问题问题1&#xff1a;修改默认工作路径启动失败问题2 找不到对应组 基础信息 操作系统信息 OS版本&#xff1a;CentOS 7 64位 内核版本&#xff1a;3.10.0 相关命令&#xff1a; uname -rcat /etc/os-rele…...

JS设计模式(4):观察者模式

JS设计模式(4):观察者模式 一、引入 在开发中&#xff0c;我们经常会遇到这样的场景&#xff1a;一个对象的状态变化需要自动通知其他对象&#xff0c;比如&#xff1a; 电商平台中&#xff0c;商品库存变化时需要通知所有订阅该商品的用户&#xff1b;新闻网站中&#xff0…...

Python 实现 Web 静态服务器(HTTP 协议)

目录 一、在本地启动 HTTP 服务器1. Windows 下安装 node.js1&#xff09;下载安装包2&#xff09;配置环境变量3&#xff09;安装镜像4&#xff09;node.js 的常用命令 2. 安装 http-server 服务3. 使用 http-server 开启服务1&#xff09;使用 http-server2&#xff09;详解 …...

毫米波雷达基础理论(3D+4D)

3D、4D毫米波雷达基础知识及厂商选型 PreView : https://mp.weixin.qq.com/s/bQkju4r6med7I3TBGJI_bQ 1. FMCW毫米波雷达基础知识 主要参考博文&#xff1a; 一文入门汽车毫米波雷达基本原理 &#xff1a;https://mp.weixin.qq.com/s/_EN7A5lKcz2Eh8dLnjE19w 毫米波雷达基础…...