防止消息丢失与消息重复——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副本的数据。

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.mzph.cn/news/117246.shtml

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

相关文章

Android 12.0 根据app包名授予app监听系统通知权限

Android 12.0 根据app包名授予app监听系统通知权限

1.概述 在12.0的系统rom产品定制化开发中，在一些产品核心开发中，第三方app需要开启系统通知权限，然后可以在app中，监听系统所有通知，来做个通知中心的功能，所以需要授权 获取系统通知的权限，然后来顺利的监听系统通知。来做系统通知的功能，首选分析下相关授权通知的功…
阅读更多...
Netty实战-实现自己的通讯框架

Netty实战-实现自己的通讯框架

通信框架功能设计 功能描述 通信框架承载了业务内部各模块之间的消息交互和服务调用&#xff0c;它的主要功能如下&#xff1a; 基于 Netty 的 NIO 通信框架&#xff0c;提供高性能的异步通信能力&#xff1b;提供消息的编解码框架&#xff0c;可以实现 POJO 的序列化和反序…
阅读更多...
小程序 wxml2canvas开发文档

小程序 wxml2canvas开发文档

wxml: <view class"share__canvas share__canvas1"><view class"share__canvas1-text draw_canvas" data-type"text" data-text"这是一段无边距文字">这是一段无边距文字</view> </view> <canvas canvas-…
阅读更多...
BurpSuite安装

BurpSuite安装

下载 BurpSuite 下载 Java17 下载后确定版本 java -version获取启动器 密钥生成器 破解 将下载的 BurpSuite、启动器、密钥生成器&#xff0c;放入同一个目录 打开 CMD 进入该目录 启动密钥生成器 java -jar burp-keygen-scz.jar开启新的CMD&#xff0c;进入该目录 启动…
阅读更多...
paddlepaddle使用实践过程中的问题记录

paddlepaddle使用实践过程中的问题记录

环境背景 python&#xff1a;3.10.11 系统&#xff1a;macOS Big Sur 11.7.10 cpu&#xff1a;Intel Core i7 2.6GHz 内存&#xff1a;16G paddle版本问题 初始安装的是的MacOS cpu 2.5.1版本&#xff0c;在python解释器中执行import paddle时报错&#xff1a; ImportError:…
阅读更多...
SpringCache配置Redis有效解决缓存击穿和缓存雪崩问题

SpringCache配置Redis有效解决缓存击穿和缓存雪崩问题

初始代码 作者参考的一篇CSDN的配置函数代码&#xff0c;实在不好意思&#xff0c;作者忘记是哪位博主了&#xff1a; /*** 设置CacheManager缓存规则* param factory* return*/Beanpublic CacheManager cacheManager(RedisConnectionFactory factory) {RedisSerializer<St…
阅读更多...
Spring Boot - 启动主要流程

Spring Boot - 启动主要流程

Spring Boot的启动主要流程可以概括为以下几个步骤&#xff1a; 加载Spring Boot配置&#xff1a; Spring Boot应用的启动从加载配置开始。Spring Boot会读取application.properties或application.yml等配置文件&#xff0c;将配置信息加载到Spring的Environment中&#xff0c;…
阅读更多...
IOS屏幕旋转监听

IOS屏幕旋转监听

IOS屏幕旋转 1.设计窗口,添加三个按钮 2.添加事件连接 3.按钮点击事件实现 先添加三个IBAction 实现IBAction 使用旋转立刻生效 -(IBAction)btnFixPortrait:(id)sender{//访问应用程序委托成员_app.mask UIInterfaceOrientationMaskPortrait;//设置窗口旋转属性[self setN…
阅读更多...
企业安全—SDL概述篇

企业安全—SDL概述篇

0x00 前言 众所周知&#xff0c;从源头开始就开发安全的代码&#xff0c;比产品已经成型之后付出的代价要小很多&#xff0c;也就是一直在说的安全左移的概念。最好就是从一开始&#xff0c;大家就用最安全的代码&#xff0c;或者是框架&#xff0c;那么开发出来的产品必然会减…
阅读更多...
git rebase -i 详解

git rebase -i 详解

git rebase 命令简介 git rebase命令允许我们轻松地更改一系列提交&#xff0c;修改存储库的历史记录。我们可以重新排序、编辑或合并提交。一般常用git rebase来合并当前分支的多个commit记录&#xff08;压缩&#xff09;以及避免出现分支的交叉合并&#xff08;变基&#x…
阅读更多...
ChatGPT课件汇总介绍

ChatGPT课件汇总介绍

第二节：有效管理 Token，充分发挥 ChatGPT 的能力 OpenAI 官方计算token的测试地址：https://platform.openai.com/tokenizer 第三节：探索ChatGPT在不同领域的创新应用 1、小说撰写 1、我希望你能作为一个小说家。我会给你一个主题，请写出有创意的、吸引人的故事，能够长时…
阅读更多...
MongoDB 学习笔记（基础）

MongoDB 学习笔记（基础）

概论 出现背景&#xff1a;MongoDB 是文档型数据库&#xff0c;由于传统的关系型数据库&#xff08;如 MySQL&#xff09;&#xff0c;在数据操作的“三高”需求以及应对 web 的网站需求面前显得有些吃力&#xff0c;在此环境下 MongoDB 出世了 三高需求&#xff1a; (1) 对数…
阅读更多...
进程的优先级与LAMP项目部署实战

进程的优先级与LAMP项目部署实战

一、进程的优先级&#xff08;扩展&#xff09; 1、什么是进程的优先级 Linux是一个多用户、多任务的操作系统&#xff0c;系统中通常运行着非常多的进程。哪些进程先运行&#xff0c;哪些进程后运行&#xff0c;就由进程优先级来控制 思考&#xff1a;什么时候需要用到进程…
阅读更多...
力扣-python-两数相加

力扣-python-两数相加

题解 1: # Definition for singly-linked list. # class ListNode(object): # def __init__(self, val0, nextNone): # self.val val # self.next nextclass Solution(object):def addTwoNumbers(self, l1, l2):""":type l1: ListNode:t…
阅读更多...
【MATLAB源码-第56期】基于WOA白鲸优化算法和PSO粒子群优化算法的三维路径规划对比。

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

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

2023-10-17 LeetCode每日一题（倍数求和）

2023-10-17每日一题 一、题目编号 2652. 倍数求和二、题目链接 点击跳转到题目位置 三、题目描述 给你一个正整数 n &#xff0c;请你计算在 [1&#xff0c;n] 范围内能被 3、5、7 整除的所有整数之和。 返回一个整数&#xff0c;用于表示给定范围内所有满足约束条件的数…
阅读更多...
华纳云：连接mysql出现2059错误怎么解决

华纳云：连接mysql出现2059错误怎么解决

MySQL连接错误2059通常表示MySQL服务器拒绝了连接。这种错误可能由多种原因引起&#xff0c;以下是一些可能的解决方法&#xff1a; 检查MySQL服务器是否正在运行&#xff1a; 确保MySQL服务器正在正常运行。您可以使用以下命令检查MySQL服务器的状态&#xff1a; systemctl st…
阅读更多...
Object.prototype.toString.call() 和 instanceOf 和 Array.isArray() 详解

Object.prototype.toString.call() 和 instanceOf 和 Array.isArray() 详解

解析: 理解 Object.prototype.toString.call(), instanceof, 和 Array.isArray() 是 JavaScript 中重要的类型检查工具。以下是一个较为详细的解释和示例代码&#xff0c;帮助你理解它们的工作原理和使用场景 Object.prototype.toString.call()&#xff1a; Object.prototyp…
阅读更多...
rust - 理解borrow trait

rust - 理解borrow trait

简介 borrow trait 是处理借用(即其它语言中的引用)的 trait,变量的所有权不会转移.泛型定义如下: pub trait Borrow<Borrowed: ?Sized> {/// Immutably borrows from an owned value.fn borrow(&self) -> &Borrowed; }其中包含一个 borrow(&self)的方…
阅读更多...
动手学深度学习—含并行连结的网络GoogLeNet（代码详解）

动手学深度学习—含并行连结的网络GoogLeNet（代码详解）

目录 1. Inception块3. GoogLeNet模型3. 训练模型 GoogLeNet吸收了NiN中串联网络的思想&#xff0c;并在此基础上做了改进&#xff0c;并且在2014年的ImageNet图像识别挑战赛中获得了不错的效果。 1. Inception块 GoogLeNet论文解决了多大的卷积核最合适的问题。 Inception块…
阅读更多...
最新文章

Copyright @ 2022~2023