从kafka如何保证数据一致性看通常数据一致性设计

一、前言

在数据库系统中有个概念叫事务，事务的作用是为了保证数据的一致性，意思是要么数据成功，要么数据失败，不存在数据操作了一半的情况，这就是数据的一致性。在很多系统或者组件中，很多场景都需要保证数据的一致性，有的是高度的一致性。特别是在交易系统等这样场景。有些组件的数据不一定需要高度保证数据的一致性，比如日志系统。本节从从kafka如何保证数据一致性看通常数据一致性设计。

二、kafka那些环节存在数据不一致性

  1. 数据复制：在Kafka中，数据从主节点（leader）复制到从节点（follower）的过程中，由于网络延迟、节点故障或其他原因，可能导致从节点未能及时获取或处理主节点的数据变更，从而产生数据不一致。
  2. 消息提交：Kafka中的消息提交涉及多个阶段，包括生产者发送消息、消息被写入日志、消息被复制到从节点等。如果在这个过程中发生错误或异常，可能导致消息丢失或重复，进而引发数据不一致。
  3. 消费者处理：消费者在处理消息时，如果因为某些原因（如网络中断、消费者进程崩溃等）未能成功处理消息，而消息又被重新投递给其他消费者处理，也可能导致数据不一致。
  4. 分区重新分配：在Kafka中，如果分区的leader节点发生故障，Kafka会触发分区重新分配，将leader切换到其他节点。在这个过程中，如果切换不及时或切换过程中发生错误，可能导致数据不一致。

三、kafka如何保证数据一致性（内容摘自半亩方塘立身）

我们知道Kafka架构如下，主要由 Producer、Broker、Consumer 三部分组成。一条消息从生产到消费完成这个过程，可以划分三个阶段，生产阶段、存储阶段、消费阶段。

• 生产阶段: 在这个阶段，从消息在 Producer 创建出来，经过网络传输发送到 Broker 端。

• 存储阶段: 在这个阶段，消息在 Broker 端存储，如果是集群，消息会在这个阶段被复制到其他的副本上。

• 消费阶段: 在这个阶段，Consumer 从 Broker 上拉取消息，经过网络传输发送到Consumer上。

那么如何保证消息不丢我们可以从这三部分来分析。

消息传递语义

在深度剖析消息丢失场景之前，我们先来聊聊「消息传递语义」到底是个什么玩意？

所谓的消息传递语义是 Kafka 提供的 Producer 和 Consumer 之间的消息传递过程中消息传递的保证性。主要分为三种。

作者：半亩方塘立身
链接：https://zhuanlan.zhihu.com/p/682321210
来源：知乎
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。
 

1. 首先当 Producer 向 Broker 发送数据后，会进行 commit，如果 commit 成功，由于 Replica 副本机制的存在，则意味着消息不会丢失，但是 Producer 发送数据给 Broker 后，遇到网络问题而造成通信中断，那么 Producer 就无法准确判断该消息是否已经被提交（commit），这就可能造成 at least once 语义。

2. 在 Kafka 0.11.0.0 之前， 如果 Producer 没有收到消息 commit 的响应结果，它只能重新发送消息，确保消息已经被正确的传输到 Broker，重新发送的时候会将消息再次写入日志中；而在 0.11.0.0 版本之后， Producer 支持幂等传递选项，保证重新发送不会导致消息在日志出现重复。为了实现这个, Broker 为 Producer 分配了一个ID，并通过每条消息的序列号进行去重。也支持了类似事务语义来保证将消息发送到多个 Topic 分区中，保证所有消息要么都写入成功，要么都失败，这个主要用在 Topic 之间的 exactly once 语义。 其中启用幂等传递的方法配置：enable.idempotence = true。 启用事务支持的方法配置：设置属性 transcational.id = "指定值"。

3. 从 Consumer 角度来剖析, 我们知道 Offset 是由 Consumer 自己来维护的, 如果 Consumer 收到消息后更新 Offset， 这时 Consumer 异常 crash 掉， 那么新的 Consumer 接管后再次重启消费，就会造成 at most once 语义（消息会丢，但不重复）。

4. 如果 Consumer 消费消息完成后, 再更新 Offset，如果这时 Consumer crash 掉，那么新的 Consumer 接管后重新用这个 Offset 拉取消息， 这时就会造成 at least once 语义（消息不丢，但被多次重复处理）。

总结： 默认 Kafka 提供「at least once」语义的消息传递，允许用户通过在处理消息之前保存 Offset的方式提供 「at mostonce」 语义。如果我们可以自己实现消费幂等，理想情况下这个系统的消息传递就是严格的「exactly once」, 也就是保证不丢失、且只会被精确的处理一次，但是这样是很难做到的。

接下来我们从生产阶段、存储阶段、消费阶段这几方面看下kafka如何保证消息不丢失。

生产阶段

通过深入解析Kafka消息发送过程我们知道Kafka生产者异步发送消息并返回一个Future，代表发送结果。首先需要我们获取返回结果判断是否发送成功。

// 异步发送消息，并设置回调函数 
producer.send(record, new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception exception) {if (exception != null) { System.err.println("消息发送失败: " + exception.getMessage()); } else { System.out.println("消息发送成功到主题: " + metadata.topic() + ", 分区: " + metadata.partition() + ", 偏移量: " + metadata.offset()); } } 
});

作者：半亩方塘立身
链接：https://zhuanlan.zhihu.com/p/682321210
来源：知乎
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。
 

消息队列通过最常用的请求确认机制，来保证消息的可靠传递：当你的代码调用发消息方法时，消息队列的客户端会把消息发送到 Broker，Broker 收到消息后，会给客户端返回一个确认响应，表明消息已经收到了。客户端收到响应后，完成了一次正常消息的发送。

Producer（生产者）保证消息不丢失的方法：

1. 发送确认机制：Producer可以使用Kafka的acks参数来配置发送确认机制。通过设置合适的acks参数值，Producer可以在消息发送后等待Broker的确认。确认机制提供了不同级别的可靠性保证，包括：

• acks=0：Producer在发送消息后不会等待Broker的确认，这可能导致消息丢失风险。

• acks=1：Producer在发送消息后等待Broker的确认，确保至少将消息写入到Leader副本中。

• acks=all或acks=-1：Producer在发送消息后等待Broker的确认，确保将消息写入到所有ISR（In-Sync Replicas）副本中。这提供了最高的可靠性保证。

2. 消息重试机制：Producer可以实现消息的重试机制来应对发送失败或异常情况。如果发送失败，Producer可以重新发送消息，直到成功或达到最大重试次数。重试机制可以保证消息不会因为临时的网络问题或Broker故障而丢失。

作者：半亩方塘立身
链接：https://zhuanlan.zhihu.com/p/682321210
来源：知乎
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。
 

Broker存储阶段

正常情况下，只要 Broker 在正常运行，就不会出现丢失消息的问题，但是如果 Broker 出现了故障，比如进程死掉了或者服务器宕机了，还是可能会丢失消息的。

在kafka高性能设计原理中我们了解到kafka为了提高性能用到了 Page Cache 技术.在读写磁盘日志文件时，其实操作的都是内存，然后由操作系统决定什么时候将 Page Cache 里的数据真正刷入磁盘。如果内存中数据还未刷入磁盘服务宕机了，这个时候还是会丢消息的。

为了最大程度地降低数据丢失的可能性，我们可以考虑以下方法：

  1. 持久化配置优化：可以通过调整 Kafka 的持久化配置参数来控制数据刷盘的频率，从而减少数据丢失的可能性。例如，可以降低 flush.messages 和 flush.ms 参数的值，以更频繁地刷写数据到磁盘。
  2. 副本因子增加：在 Kafka 中，可以为每个分区设置多个副本，以提高数据的可靠性。当某个 broker 发生故障时，其他副本仍然可用，可以避免数据丢失。
  3. 使用acks=all：在生产者配置中，设置 acks=all 可以确保消息在所有ISR（In-Sync Replicas）中都得到确认后才被视为发送成功。这样可以确保消息被复制到多个副本中，降低数据丢失的风险。
  4. 备份数据：定期备份 Kafka 的数据，以便在发生灾难性故障时可以进行数据恢复。

消费阶段

消费阶段采用和生产阶段类似的确认机制来保证消息的可靠传递，客户端从 Broker 拉取消息后，执行用户的消费业务逻辑，成功后，才会给 Broker 发送消费确认响应。如果 Broker 没有收到消费确认响应，下次拉消息的时候还会返回同一条消息，确保消息不会在网络传输过程中丢失，也不会因为客户端在执行消费逻辑中出错导致丢失。

  1. 自动提交位移：Consumer可以选择启用自动提交位移的功能。当Consumer成功处理一批消息后，它会自动提交当前位移，标记为已消费。这样即使Consumer发生故障，它可以使用已提交的位移来恢复并继续消费之前未处理的消息。
  2. 手动提交位移：Consumer还可以选择手动提交位移的方式。在消费一批消息后，Consumer可以显式地提交位移，以确保处理的消息被正确记录。这样可以避免重复消费和位移丢失的问题。
作者：半亩方塘立身
链接：https://zhuanlan.zhihu.com/p/682321210
来源：知乎
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。// 创建消费者实例
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);// 订阅主题
consumer.subscribe(Collections.singletonList(topic));try {while (true) {// 消费消息ConsumerRecords<String, String> records = consumer.poll(100);for (ConsumerRecord<String, String> record : records) {// 处理消息逻辑System.out.println("消费消息：Topic = " + record.topic() +", Partition = " + record.partition() +", Offset = " + record.offset() +", Key = " + record.key() +", Value = " + record.value());// 手动提交位移TopicPartition topicPartition = new TopicPartition(record.topic(), record.partition());OffsetAndMetadata offsetMetadata = new OffsetAndMetadata(record.offset() + 1);consumer.commitSync(Collections.singletonMap(topicPartition, offsetMetadata));}}
} catch (Exception e) {e.printStackTrace();
} finally {consumer.close();
}

 

 四、数据一致系统设计特点

从kafka如何保证数据一致性看通常数据一致性设计，一般保证数据一致性，需要通过成功后commit的操作，消费过程中记录小标。成功与失败的环节都记上标志。

Kafka作为一个分布式发布-订阅消息系统，其数据一致性的系统设计特点主要包括以下几个方面：

  1. 分区与副本机制：Kafka将数据分成多个分区（Partition），每个分区在集群中有多个副本（Replica）。这些副本分布在不同的Broker上，以实现数据的冗余备份和高可用性。当某个Broker发生故障时，其他Broker上的副本可以接管服务，保证数据的持续可用。
  2. ISR（In-Sync Replicas）机制：ISR是Kafka中用于维护数据一致性的重要机制。它包含所有与Leader保持同步的副本。当ISR中的副本数量不足时，Kafka会暂停写入操作，以防止数据不一致。只有当ISR中的副本数量恢复到一定数量时，才会恢复写入操作。
  3. 消息提交确认：生产者发送消息到Kafka时，需要等待消息被写入ISR中的副本并得到确认，以确保消息被成功存储。同时，消费者在处理消息时也需要定期提交偏移量（Offset），以便在发生故障时能够从正确的位置继续消费。
  4. 原子性操作：Kafka保证消息在分区内的顺序性和原子性。这意味着在同一分区内的消息会按照发送的顺序被消费，且不会被其他消息插入打断。这有助于保证数据的一致性和正确性。
  5. 容错处理：当Kafka集群中的节点发生故障时，Kafka会自动进行故障转移和恢复操作。这包括从ISR中选择新的Leader、重新同步数据等，以确保数据的持续可用和一致性。

总之，Kafka通过分区与副本机制、ISR机制、消息提交确认、原子性操作和容错处理等手段，确保了其数据一致性的系统设计特点。这些设计使得Kafka能够在分布式环境中实现高吞吐量、持久化存储、可扩展性和高可靠性等特性，从而满足各种复杂场景下的数据一致性需求。

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

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

相关文章

计算机网络——18无连接传输UDP

计算机网络——18无连接传输UDP

无连接传输UDP UDP “尽力而为的”服务&#xff0c;报文段可能 丢失送到应用进程的报文段乱序 无连接 UDP发送端和接收端之间没有握手每个UDP报文段都被独立的处理 UDP被用于 流媒体DNSSNMP 在UDP上实现可靠传输 在应用层增加可靠性应用特定的差错格式 UDP&#xff1a;用户…
阅读更多...
Positive SSL 证书介绍

Positive SSL 证书介绍

Positive SSL 是一种受欢迎的 SSL 证书&#xff0c;提供了卓越的安全性、性价比和品牌信任。以下是对 Positive SSL 在这些方面的简要介绍&#xff1a; 1. 安全性&#xff1a; Positive SSL 证书采用强大的加密技术&#xff0c;确保网站和用户之间的数据传输是安全的。它使用…
阅读更多...
turn服务器debug

turn服务器debug

turn服务器正常能连通的调用堆栈 turn_port.cc AddRequestAuthInfo check 崩溃 有问题的turn msg type是259 request type 是3 用不了的turn 服务器turnmessage type 275
阅读更多...
Mybatis源码分析

Mybatis源码分析

Mybatis源码分析 第一章、回顾 1. 课程中工具的版本 1. JDK8 2. IDEA2018.3 3. Maven3.5.3 4. MySQL 5.1.48 --> MySQL 5Mybatis 3.4.62. Mybatis开发的简单回顾 1. Mybatis做什么&#xff1f;Mybatis是一个ORM类型框架&#xff0c;解决的数据库访问和操作的问题&#xf…
阅读更多...
Github：分享一款开源的跨平台多功能远程控制和监控工具Spark

Github：分享一款开源的跨平台多功能远程控制和监控工具Spark

目录 1、设备列表及操作列表 2、登录机器 Shell 执行命令 3、桌面远程访问 4、远程设备的进程管理 5、远程设备文件管理 今天要给大家推荐一个开源的跨平台多功能远程控制和监控工具&#xff1a;Spark。 目前该项目处于不维护状态&#xff0c;大家可以自己根据需要进行扩…
阅读更多...
代码随想录 Leetcode763. 划分字母区间

代码随想录 Leetcode763. 划分字母区间

题目&#xff1a; 代码(首刷看解析 2024年2月18日&#xff09;&#xff1a; class Solution { public:vector<int> partitionLabels(string s) {int hash[27] {0};for (int i 0; i < s.size(); i) {hash[s[i] - a] i;}vector<int> res;int left 0;int righ…
阅读更多...
PyTorch使用Tricks：学习率衰减 ！！

PyTorch使用Tricks：学习率衰减 ！！

文章目录 前言 1、指数衰减 2、固定步长衰减 3、多步长衰减 4、余弦退火衰减 5、自适应学习率衰减 6、自定义函数实现学习率调整&#xff1a;不同层不同的学习率 前言 在训练神经网络时&#xff0c;如果学习率过大&#xff0c;优化算法可能会在最优解附近震荡而无法收敛&#x…
阅读更多...
【开源】SpringBoot框架开发智能教学资源库系统

【开源】SpringBoot框架开发智能教学资源库系统

目录 一、摘要1.1 项目介绍1.2 项目录屏 二、功能模块2.1 数据中心模块2.2 课程档案模块2.3 课程资源模块2.4 课程作业模块2.5 课程评价模块 三、系统设计3.1 用例设计3.2 数据库设计3.2.1 课程档案表3.2.2 课程资源表3.2.3 课程作业表3.2.4 课程评价表 四、系统展示五、核心代…
阅读更多...
超级楼梯（动态规划）

超级楼梯（动态规划）

#include <iostream> using namespace std; int main (){int n;cin >> n;int f[41];f[1] 0;f[2] 1;f[3] 2;for (int i 4;i < 42;i){f[i] f[i-1] f[i-2];}while (n--){int stair;cin>> stair;cout << f[stair]<<endl;}return 0; }
阅读更多...
【ChatIE】论文解读：Zero-Shot Information Extraction via Chatting with ChatGPT

【ChatIE】论文解读：Zero-Shot Information Extraction via Chatting with ChatGPT

文章目录 介绍ChatIEEntity-Relation Triple Extration (RE)Named Entity Recognition (NER)Event Extraction (EE) 实验结果结论 论文&#xff1a;Zero-Shot Information Extraction via Chatting with ChatGPT 作者&#xff1a;Xiang Wei, Xingyu Cui, Ning Cheng, Xiaobin W…
阅读更多...
嵌入式——EEPROM（AT24C02）

嵌入式——EEPROM（AT24C02）

目录 一、初识AT24C02 1. 介绍 2. 引脚功能 补&#xff1a; 二、AT24C02组成 1. 存储结构 2. AT24C02通讯地址 3. AT24C02寻址方式 &#xff08;1&#xff09;芯片寻址 &#xff08;2&#xff09;片内子地址寻址 三、AT24C02读写时序 1. 写操作 &#xff08;1&…
阅读更多...
Linux|centos7下的编译|ffmpeg的二进制安装

Linux|centos7下的编译|ffmpeg的二进制安装

Windows版本的ffmpeg&#xff1a; ###注意&#xff0c;高版本可能必须要windows10以及以上才支持&#xff0c;win7估计是用不了的 下载地址&#xff1a;Builds - CODEX FFMPEG gyan.dev 或者这个下载地址&#xff1a;https://github.com/BtbN/FFmpeg-Builds/releases 这两个…
阅读更多...
ClickHouse--12-可视化工具操作

ClickHouse--12-可视化工具操作

提示&#xff1a;文章写完后&#xff0c;目录可以自动生成&#xff0c;如何生成可参考右边的帮助文档 文章目录 可视化工具操作1 tabixhttp://ui.tabix.io/ 2 DBeaverhttps://dbeaver.io/download/ 可视化工具操作 1 tabix tabix 支持通过浏览器直接连接 ClickHouse&#xff…
阅读更多...
【制作100个unity游戏之25】3D背包、库存、制作、快捷栏、存储系统、砍伐树木获取资源、随机战利品宝箱12（附带项目源码）

【制作100个unity游戏之25】3D背包、库存、制作、快捷栏、存储系统、砍伐树木获取资源、随机战利品宝箱12（附带项目源码）

效果演示 文章目录 效果演示系列目录前言悬停显示物品详情源码完结 系列目录 前言 欢迎来到【制作100个Unity游戏】系列&#xff01;本系列将引导您一步步学习如何使用Unity开发各种类型的游戏。在这第25篇中&#xff0c;我们将探索如何用unity制作一个3D背包、库存、制作、快…
阅读更多...
11.【CPP】模版（深入理解模版的实例化，从编译链接的原理理解模版为何无法分离编译）

11.【CPP】模版（深入理解模版的实例化，从编译链接的原理理解模版为何无法分离编译）

非类型模版参数 1.模版参数分为类型模版参数和非类型模版参数&#xff0c;非类型模版参数一般都是整形常量&#xff08;整形&#xff1a;size_t,int,char等&#xff09; 2.浮点数、类对象以及字符串是不允许作为非类型模版参数的。非类型模版的参数必须在编译的时候就能确定结…
阅读更多...
leetcode hot100 拆分整数

leetcode hot100 拆分整数

在本题目中&#xff0c;我们需要拆分一个整数n&#xff0c;让其拆分的整数积最大。因为每拆分一次都和之前上一次拆分有关系&#xff0c;比如拆分6可以拆成2x4&#xff0c;还可以拆成2x2x2&#xff0c;那么我们可以采用动态规划来做。 首先确定dp数组的含义&#xff0c;这里dp…
阅读更多...
第13章 网络 Page744~746 asio核心类 ip::tcp::endPoint

第13章 网络 Page744~746 asio核心类 ip::tcp::endPoint

2. ip::tcp::endpoint ip::tcp::socket用于连接TCP服务端的 async_connect()方法的第一个入参是const endpoint_type& peer_endpoint. 此处的类型 endpoint_type 是 ip::tcp::endpoint 在 在 ip::tcp::socket 类内部的一个别名。 libucurl 库采用字符串URL表达目标的地…
阅读更多...
LeetCode 100题目（python版本）待续...

LeetCode 100题目（python版本）待续...

一.哈希 1.两数之和 题目 给定一个整数数组 nums 和一个整数目标值 target&#xff0c;请你在该数组中找出 和为目标值 target 的那 两个 整数&#xff0c;并返回它们的数组下标。 你可以假设每种输入只会对应一个答案。但是&#xff0c;数组中同一个元素在答案里不能重复…
阅读更多...
Acwing---846. 树的重心

Acwing---846. 树的重心

树的重心 1.题目2.基本思想3.代码实现 1.题目 给定一颗树&#xff0c;树中包含 n n n 个结点&#xff08;编号 1 ∼ n 1∼n 1∼n&#xff09;和 n − 1 n−1 n−1 条无向边。 请你找到树的重心&#xff0c;并输出将重心删除后&#xff0c;剩余各个连通块中点数的最大值。 …
阅读更多...
【C Primer Plus第六版 学习笔记】 第十七章 高级数据表示

【C Primer Plus第六版 学习笔记】 第十七章 高级数据表示

有基础&#xff0c;进阶用&#xff0c;个人查漏补缺 链表&#xff1a;假设要编写一个程序&#xff0c;让用户输入一年内看过的所有电影&#xff0c;要储存每部影片的片名和评级。 #include <stdio.h> #include <stdlib.h> /* 提供malloc()的原型 */ #include <s…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023