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

系列文章目录

上手第一关，手把手教你安装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副本的数据。

ISR_41">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.niftyadmin.cn/n/5118732.html

相关文章

李沐——论文阅读——VIT(VIsionTransformer)

李沐——论文阅读——VIT(VIsionTransformer)

一、终极结论&#xff1a; 如果在足够多的数据上面去做预训练&#xff0c;那么&#xff0c;我们也可以不用 卷积神经网络&#xff0c;而是直接用 自然语言处理那边搬过来的 Transformer&#xff0c;也能够把视觉问题解决的很好 &#xff08;tips&#xff1a;paperswithcode.co…
阅读更多...
Java反射获取内部类方法

Java反射获取内部类方法

Java反射获取内部类方法 结论一、案例准备二、测试方法&#xff1a;使用反射获取类的成员内部类和方法具体操作具体操作&#xff08;使用getDeclaredClasses&#xff09; 结论 Java 通过反射可以获得内部类&#xff0c;包括内部类属性信息和方法。 一、案例准备 创建了一个类…
阅读更多...
微信小程序开发之投票管理及小程序UI的使用

微信小程序开发之投票管理及小程序UI的使用

目录 一、小程序UI 1.讲述 2. 介绍vantWeapp 3. 使用vantWeapp 安装 构建 依赖 引用 二、后端 1. 后端实体对象 2. 后端接口 3. 实现类 4. 请求处理类 三、前端 1. 定义路径 2. 页面引用 3. 页面 4. 页面美化 5. 数据 6. 效果展示 一、小程序UI 1.讲述 小…
阅读更多...
【MATLAB源码-第56期】基于WOA白鲸优化算法和PSO粒子群优化算法的三维路径规划对比。

【MATLAB源码-第56期】基于WOA白鲸优化算法和PSO粒子群优化算法的三维路径规划对比。

操作环境&#xff1a; MATLAB 2022a 1、算法描述 1.粒子群算法&#xff08;Particle Swarm Optimization&#xff0c;简称PSO&#xff09;是一种模拟鸟群觅食行为的启发式优化方法。以下是其详细描述&#xff1a; 基本思想&#xff1a; 鸟群在寻找食物时&#xff0c;每只鸟都…
阅读更多...
docker（3） dockerCompose

docker（3） dockerCompose

Docker Compose通过一个单独的docker-compose.yml 模板文件&#xff08;YAML 格式&#xff09;来定义一组相关联的应用容器&#xff0c;帮助我们实现多个相互关联的Docker容器的快速部署。 先将之前的容器全部删除 运行命令 注意需在compose文件同目录执行命令&#xff0c;否…
阅读更多...
RHCE---流程控制之条件判断

RHCE---流程控制之条件判断

文章目录 目录 文章目录 前言 一.if 语句语法 双分支结构 多分支结构 案列分析 总结 前言 命令连接符号中&&以及|| 有各自的判断条件&#xff0c;这是在shell基础中了解到的条件判断&#xff0c;而在脚本编写中&#xff0c;shell脚本有着自己的流程控制之条件判断…
阅读更多...
Standalone---基础认知

Standalone---基础认知

参考&#xff1a;doc/README.standalone The functions are exported by U-Boot via a jump table. 跳转表的初始函数jumptable_init()在common/exports.c文件中实现跳转表结构体在include/exports.h文件中定义 // u-boot/include/asm-generic/global_data.h typedef struct gl…
阅读更多...
【机器学习】数据均衡学习笔记

【机器学习】数据均衡学习笔记

文章目录 序言1. 样本不均衡2. 样本不均衡的影响以及样本均衡的意义3. 什么时候需要进行样本均衡/数据均衡4. 数据不均衡的解决办法 序言 数据集制作过程中需要关注样本均衡问题&#xff0c;学习笔记&#xff0c;简单记录 1. 样本不均衡 分类任务中不同类别样本数差别很大的…
阅读更多...
最新文章

Copyright @ 2022~2023