防止消息丢失与消息重复——Kafka可靠性分析及优化实践

news2024/11/18 6:01:17

系列文章目录

上手第一关，手把手教你安装kafka与可视化工具kafka-eagle
Kafka是什么，以及如何使用SpringBoot对接Kafka
架构必备能力——kafka的选型对比及应用场景
Kafka存取原理与实现分析，打破面试难关

防止消息丢失与消息重复——Kafka可靠性分析及优化实践

在这里插入图片描述
在上一章内容中，我们解析了Kafka在读写层面上的原理，介绍了很多Kafka在读出与写入时的各种设计，初步理解了Kafka大吞吐量的原因，本期我们将带领大家从另一个角度，即从可靠性方面来分析Kafka的机制与原理

📕作者简介：战斧，从事金融IT行业，有着多年一线开发、架构经验；爱好广泛，乐于分享，致力于创作更多高质量内容
📗本文收录于 kafka 专栏，有需要者，可直接订阅专栏实时获取更新
📘高质量专栏 云原生、RabbitMQ、Spring全家桶 等仍在更新，欢迎指导
📙Zookeeper Redis dubbo docker netty等诸多框架，以及架构与分布式专题即将上线，敬请期待

一、可靠性的考量角度

其实我们在《RabbitMQ 能保证消息可靠性吗》 一文中，我们已经阐述了对于MQ类主键，可靠性应该从哪些角度去判断。总结下来其实就是：

  • 消息不会意外丢失
  • 消息不会重复传递

那么本次我们也将从这两方面来看看 Kafka 都做了哪些工作来提升可靠性。

二、分区副本

1. 分区副本的含义

在这里插入图片描述

我们之前了解到：一个Kafka的主题被分为了若干个分区，每个分区都是一个有序的消息队列。如上图，我们就把topicA分成了1\2\3\4 四个分区（实线圆柱），但是我们还看到了更多的“虚线圆柱”，这些其实是 1\2\3\4 分区的副本，在Kafka中，每个分区都有多个副本。副本是指一个分区在其他Broker上的备份。副本的作用是提高消息的可靠性和容错性，一旦某个Broker宕机，其他Broker上的副本就可以接替宕机的Broker继续提供服务。

我们可以看到不同的Broker之间。互相存储着对方的副本分区，比如Partition1 存在Broker1 上，但Partition1 的副本可以放在Broker2 ，同理 Partition2 的副本也可以放在Broker1 ，如果我们专注于一个分区，那么其情况如下：

在这里插入图片描述

为了区分，我们一般把“实线圆柱”的分区称为 Leader，“虚线圆柱”的分区称为Follower。Leader副本负责读写请求，而Follower副本则只负责复制Leader副本的数据。

2. AR 与 ISR机制

上面我们讲了副本分区的Follower，所有副本的合集统称为AR(Assigned Replicas)，但是不同副本和Leader到底一致性有多高呢？会不会出现有的副本同步得及时，有的副本因为网络原因同步得很慢呢？ 这里又引出了一个重要的概念叫做ISR（In-Sync Replica）机制。

ISR，是一个机制，也代表着一个同步合集，顾名思义，它包含着所有处于同步状态的副本。当一个副本和Leader副本的差距超过一定程度时，这个副本就会被认为是不同步的，不再被加入到ISR中。也因此，Kafka中的 ISR 并不是一直不变的

在这里插入图片描述

那么，既然ISR是动态的，那哪些副本会被包含在ISR中呢？

其实，其主要依据就是 副本需要保证能够及时地接收并复制Leader副本的消息，也就是需要保证与leader副本的消息同步延迟在一定的时间范围内（默认情况下是10秒钟，由参数 replica.lag.time.max.ms 控制）。

在这里插入图片描述

换而言之，因为分区与ISR机制，我们的消息一旦被Kafka 接收后，就会复制多份并很快落盘。这意味着，即使某一台Broker节点宕机乃至硬盘损毁，也不会导致数据丢失。

三、ACKS设置

如果说备份机制是保障消息不会在Kafka服务器丢失，那么消息丢失的另一个重要原因就是消息在发送中丢失。

这种场景下，我们就需要利用消息确认机制了，此时我们也会利用到ISR，比如我们在发送消息时，可以通过设置ACK的值，来决定同步的情况：

  • acks=0，如果设置为零，则生产者根本不会等待来自服务器的任何确认。该记录将立即添加到套接字缓冲区，并被视为已发送。在这种情况下，不能保证服务器已经收到记录，重试配置也不会生效（因为客户端通常不会知道任何故障）。为每条记录返回的偏移量将始终设置为-1。

  • acks=1，这意味着领导者会将记录写入其本地日志，但不会等待所有追随者的完全确认。在这种情况下，如果领导者在确认记录后立即失败，但在追随者复制之前，记录将丢失。

  • acks=all，这意味着leader将等待所有ISR来确认记录。这保证了只要至少有一个副本处于ISR内，记录就不会丢失。这是最有力的保证。这相当于acks=-1的设置。

当我们设置acks=all 或者 -1 的时候， broker 的 min.insync.replicas 参数起作用（一个典型的场景是，topic 有 3个 副本，客户端设置 acks = -1，服务端设置 topic level 的 min.insync.replicas = 2，这样至少有 2 个副本写入后，broker 才会返回；但是如果 topic 只有 1 个副本，而 acks = all，min.insync.replicas = 2，就会报 NOT_ENOUGH_REPLICAS 错误）；

在这里插入图片描述

四、重试机制

发消息不可能万无一失，当Kafka在发送或接收消息时发生错误时，可以通过重试来解决这些问题，提高系统的可靠性。这点不用多说，那么在生产端我们可以配置哪些重试的参数呢？

  • retries：重试次数，默认为0，表示不启用重试机制，但建议开启。

  • retry.backoff.ms：每次重试的时间间隔，默认为100ms。

  • retry.buffer.records：每个分区的缓冲区中可以存储的最大重试消息数。

在这里插入图片描述

在实际应用中，一些小故障是可以通过重试来解决的，但是重试次数过多也会增加网络通信的负担，甚至会导致消息堵塞。所以建议将其设置为1或2，这样可以在第一次发送失败后进行重试，从而提高消息的可靠性。但是，如果网络状况很差，或者需要处理重要的消息，可以适当增加retries的值。

五、幂等性设计

如果你仔细观察上面的ACKS的设置，相信你会发现，这并不完美：如果你将ACKs设置为-1（all），可以保证Producer到Server之间不会丢失数据，即At Least Once(最少一次)，但不能保证数据不重复。而如果你将ACKs级别设置为0（不需要等写log），则可以保证生产者每条消息只会被发送一次，即At Most Once（最多一次），但不能保证不会发生数据丢失。

难道就没有一种既不会丢失，也不会重复的方案吗？其实是有的，这个时候我们可以使用 ‘ack = all’ + 幂等性来解决，而开启幂等性 ，即设置 enable.idempotence = true

请注意，启用幂等性要求
①max.in.flight.requests.per.connection小于或等于5（为任何允许的值保留消息顺序），
②retries重试次数大于0，
③acks必须为“all”。
PS：我发现不少文章把开启幂等性参数写成 enable.idompotence，不知是笔误还是什么原因

开启幂等性的意义在于生产者将确保在流中每条消息正好只会发送一次，那么它是怎么实现的呢？Kafka生产者的幂等性算法主要包括以下三个方面的实现：

  1. 生产者编号： Producer在初始化的时候（只有初始化的时候会随机生成PID）会被分配一个PID（producerid），
  2. 分区编号：不同的分区有自己的paritionid（即分区号），
  3. 序列号：发往同一Partition的消息会附带Sequence Number（即发送数据的编号，代表着向分区发送的第几条消息），这样<PID, PartitionID, SeqNumber>就相当于构成了一个主键。Broker端会对<PID, PartitionID, SeqNumber>做缓存，当具有相同主键的消息提交时，Broker只会持久化一条。

六、消费偏移量

前面说了很多，都是生产者与Kafka的，其实对于消费者，Kakfa也有同步提交偏移量的设计。在 Kafka 的消费者 API 中，同步提交偏移量可以通过设置 enable.auto.commit 参数为 false ，然后在消费者应用中手动控制提交偏移量的时机来实现。

具体来说，可以使用 commitSync() 方法提交偏移量，该方法会一直阻塞直到提交偏移量成功，或抛出异常。示例代码如下：

Properties props = new Properties();
// 设置 Kafka 集群地址等配置信息
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("enable.auto.commit", "false"); // 禁止自动提交偏移量
props.put("auto.offset.reset", "earliest");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("test-topic"));

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            // 处理消息
        }
        consumer.commitSync(); // 手动提交偏移量
    }
} finally {
    consumer.close();
}

通过这种方式，可以确保消息在被消费者处理后再提交偏移量，从而避免了消息丢失或者重复处理的问题。同时，手动提交偏移量也提高了消费者的可控性，可以根据自身情况自由地设置提交偏移量的时机。

七、可靠性不足分析

尽管我们在上面讲了一些，Kafka为了实现可靠性而做的设计，一般情况下，这种程度的可靠性足以应付了。但在实际应用过程中，Kafka仍然可能会面临以下几个可靠性问题：

  1. 生产者重复发送：尽管开启了幂等性，但不要忘记幂等性设置仅表示生产者对同一个分区的消息的写入是有序的、幂等的，如果producer挂了，重启之后，producer会重新生成producerid，此时幂等性校验就不准了。
  2. 消费者重复消费：如果消费者在提交偏移量前宕机了，将导致Kafka认为该消息没有被消费，在消费者重启后，又会消费该消息，导致重复消费。

这些情况，Kafka自身已经无法解决。我们的解决策略只能契合在我们的业务处理上，目前一个通用的方案是全局性ID+生产/消费两端校验：
在这里插入图片描述

我们可以先根据业务对消息进行去重，然后使用诸如雪花算法等方案为每一条去重后的消息生产全局性的唯一ID，并在发送和消费之前在redis或其他较快的存储件中进行标记，这样当发生重复发送/消费时，就能及时发现了。此时你可以选择放弃本次发送/消费，也可以将该异常情况上报，由人工来进行检查与处理

总结

本次我们对Kafka的可靠性进行了分析和优化实践。一般来说，我们可以通过设置acks、开启幂等性，消费端手动提交偏移量等方式来保证可靠性，也足以应付大部分场景。而且实际应用过程中，还可以配合全局ID等手段完善可靠场景。当然，架构服务于业务需求，所以最终还是需要结合具体的业务需求和场景来选择合适的部署方式和配置参数，在后面我们还会继续进行Kafka的深入解析，如果你对此有兴趣，可以直接订阅本 kafka 专栏

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

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

相关文章

ToDesk等远程软件连接主机无法更改分辨率 - 解决方案

ToDesk等远程软件连接主机无法更改分辨率 - 解决方案

问题 使用ToDesk等远程软件连接自己的Linux或Windows主机时&#xff0c;若主机已连接显示器&#xff0c;则可通过系统设置更改显示分辨率。但如果主机没有连接显示器或显示器的电源关闭&#xff0c;则无法正常调整分辨率。下文介绍解决方案。 解决方案 方案1&#xff1a;连接…
阅读更多...
多跳推理真的可解释吗？10.24

多跳推理真的可解释吗？10.24

多跳推理真的可解释吗 摘要1 引言2 相关工作2.1 多跳推理2.2 基于规则的推理2.3 可解释性评估 3 基础知识4 基准测试4.1 数据集构建4.2 评估框架4.3 近似可解释性评分4.4 Benchmark with Manual Annotation4.5 使用挖掘规则的基准 实验 摘要 近年来&#xff0c;多跳推理在获取…
阅读更多...
BUUCTF wireshark 1

BUUCTF wireshark 1

BUUCTF:https://buuoj.cn/challenges 题目描述&#xff1a; 黑客通过wireshark抓到管理员登陆网站的一段流量包&#xff08;管理员的密码即是答案) 密文&#xff1a; 下载附件&#xff0c;解压后得到一个.pcap文件。 解题思路&#xff1a; 1、双击文件&#xff0c;在wires…
阅读更多...
测试用例的设计方法（全）：边界值分析方法

测试用例的设计方法（全）：边界值分析方法

一.方法简介 1.定义&#xff1a;边界值分析法就是对输入或输出的边界值进行测试的一种黑盒测试方法。通常边界值分析法是作为对等价类划分法的补充&#xff0c;这种情况下&#xff0c;其测试用例来自等价类的边界。 2.与等价划分的区别 1)边界值分析不是从某等价类中随便挑…
阅读更多...
Mysql主从集群同步延迟问题怎么解决

Mysql主从集群同步延迟问题怎么解决

主从复制 复制过程分为几个步骤&#xff1a; 主库的更新事件(update、insert、delete)被写到binlog 从库发起连接&#xff0c;连接到主库。 此时主库创建一个 binlog dump thread&#xff0c;把 binlog 的内容发送到从库。 从库启动之后&#xff0c;创建一个 I/O 线程&#xff…
阅读更多...
用VLOOKUP快速合并两个表格

用VLOOKUP快速合并两个表格

一、前言 上周五微信收到运营提过来的需求&#xff0c;第一句话&#xff1a;帮我提取一下1号门店的库存数据&#xff0c;马上登录系统下载一份库存数据给到他然后专心读代码&#xff0c;过一会微信第二句话&#xff1a;帮我提取一下1号门店商品半年/一年的销量数据&#xff0c…
阅读更多...
常用linux命令 linux_cmd_sheet

常用linux命令 linux_cmd_sheet

查看文件大小 ls -al 显示每个文件的kb大小 查看系统日志 dmesg -T | tail 在 top 命令中&#xff0c;RES 和 VIRT&#xff08;或者 total-vm&#xff09;是用来表示进程内存使用的两个不同指标&#xff0c;它们之间有以下区别&#xff1a; RES&#xff08;Resident Set Size…
阅读更多...
使用ruoyi框架遇到的问题修改记录

使用ruoyi框架遇到的问题修改记录

使用ruoyi框架遇到的问题修改记录 文章目录 使用ruoyi框架遇到的问题修改记录上传后文件名改变上传时设置单多文件及其他选项附件显示文件名&#xff0c;点击下载附件直接显示图片表格固定列查询数据库作为下拉选项值字典使用加入json递归注解&#xff0c;防止无限递归内存溢出…
阅读更多...
Zabbix安装与部署

Zabbix安装与部署

前言 Zabbix是一个开源的网络监控和系统监控解决方案&#xff0c;用于监控服务器、网络设备、应用程序和服务。它基于客户端-服务器体系结构&#xff0c;使用多种监控选项来监控不同类型的设备和应用程序。Zabbix支持数据收集、处理和存储&#xff0c;以及报警和可视化等功能。…
阅读更多...
VESTA软件下载

VESTA软件下载

1.进入官网添加链接描述 2.下滑找到对应版本 3.解压 4.找到.exe文件&#xff08;不用安装&#xff09;
阅读更多...
JSX 列表渲染

JSX 列表渲染

学习目标&#xff1a; 能够在 JSX 中实现列表渲染 页面的构建离不开重复的列表结构&#xff0c;比如歌曲列表&#xff0c;商品列表等等&#xff0c;Vue 中用的式 v-for 做到这一点&#xff0c;react 中又该如何实现呢&#xff1f; 实现&#xff1a; 使用数组的 map 方法 案例: …
阅读更多...
【数据科学赛】2023年亚太眼科学会大数据竞赛 #$15000 #阿里天池 #分类

【数据科学赛】2023年亚太眼科学会大数据竞赛 #$15000 #阿里天池 #分类

CompHub[1] 最新的比赛会第一时间在群里通知&#xff0c;欢迎加群交流比赛经验&#xff01;&#xff08;公众号回复“加群”即可&#xff09; 根据比赛主页[2](文末阅读原文)&#xff0c;使用AI辅助生成 大赛概况 2023年亚太眼科学会大数据竞赛由亚太眼科学会&#xff08;Asia…
阅读更多...
Xilinx FFT使用说明和测试

Xilinx FFT使用说明和测试

Xilinx FFT使用说明和测试 1 IP接口信号2 IP基本配置3 IP功能测试 本文主要介绍Xilinx FFT IP的使用方法 1 IP接口信号 FFT用于计算N点的DFT或者IDFT&#xff0c;N为2m&#xff0c;其中m2~16。IP的输入输出接口如下表所示&#xff0c;主要包括时钟、复位信号&#xff0c;输入的…
阅读更多...
ApowerREC v1.2.7.10(多功能屏幕录屏工具)

ApowerREC v1.2.7.10(多功能屏幕录屏工具)

ApowerREC是一款功能强大的屏幕录制软件&#xff0c;主要特点如下&#xff1a; 支持音画同步录制&#xff1a;可以录制电脑桌面操作、在线会议、娱乐视频等所有活动。提供多种录制模式&#xff1a;包括全屏录制、区域录制、画中画等多种录制视频模式&#xff0c;同时也可以支持…
阅读更多...
05、Python -- 爬取ts文件格式视频思路

05、Python -- 爬取ts文件格式视频思路

目录 第一步：爬取一段5秒视频找url代码结果第二步：下载整个视频的所有片段代码：结果：第三步：合成视频安装模块代码：结果简洁代码代码：结果：最终代码简洁前代码简洁后代码思路： 1、爬取视频，但是每次只能爬取一段5秒的视频。 2、一个视频有很多秒，所以需要爬取很多片…
阅读更多...
11 结构型模式- 代理模式

11 结构型模式- 代理模式

结构性模式一共包括七种&#xff1a; 代理模式、桥接模式、装饰者模式、适配器模式、门面(外观)模式、组合模式、和享元模式。 1 代理模式介绍 软件开发中的代理&#xff1a; 代理模式中引入了一个新的代理对象,代理对象在客户端对象和目标对象之间起到了中介的作用,它去掉客…
阅读更多...
分享个包含各省、市、区的编码数据的在线静态资源脚本

分享个包含各省、市、区的编码数据的在线静态资源脚本

在翻《SpringBootVue3》——十三尼克陈作者的大型前后端分离项目实战里面&#xff0c;在看到地址管理的部分时&#xff0c;发现了该作者记录有一个静态的地址资源脚本 这里做个记录&#xff0c;打点 一、引入js <script src"https://s.yezgea02.com/1641120061385/td…
阅读更多...
Python----range方法（函数）

Python----range方法（函数）

range 英 /reɪndʒ/ n. &#xff08;变动或浮动的&#xff09;范围&#xff0c;界限&#xff1b;视觉&#xff08;或听觉&#xff09;范围&#xff1b;v. &#xff08;在一定的范围内&#xff09;变化&#xff0c;变动&#xff1b;&#xff08;按一定位置或顺序&#x…
阅读更多...
如何通过PAM禁止部分用户登录

如何通过PAM禁止部分用户登录

如何通过 PAM 限制对 SSH 服务的根访问 如题。客户提出这样一个需求&#xff1a;限制和允许部分账号的SSH登录&#xff0c;限制名单可调。乍一看&#xff0c;这需求完全不合理啊&#xff1f;这又要改多少代码&#xff1f;但——PAM从脑海中一闪而过&#xff0c;想到一个办法&a…
阅读更多...
Axi接口的DDR3：参数，时序，握手机制

Axi接口的DDR3：参数，时序，握手机制

参考 AXI总线的Burst Type以及地址计算 | WRAP到底是怎么一回事&#xff1f;_axi wrap-CSDN博客 还有官方手册&#xff0c;名字太长想起来再写。 Transaction/Burst/Transfer/Beat Transaction指一次传输事务&#xff0c;实际上包括了address phase, data phase与response ph…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023