kafka如何保证消息不丢？

概述

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

图片

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

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

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

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

消息传递语义

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

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

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

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

  4. 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()); } } 
});

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

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

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

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可以显式地提交位移，以确保处理的消息被正确记录。这样可以避免重复消费和位移丢失的问题。

下面是手动提交位移的例子：

// 创建消费者实例
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();
}

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

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

相关文章

【JavaEE】_JavaScript（Web API）

【JavaEE】_JavaScript（Web API）

目录 1. DOM 1.1 DOM基本概念 1.2 DOM树 2. 选中页面元素 2.1 querySelector 2.2 querySelectorAll 3. 事件 3.1 基本概念 3.2 事件的三要素 3.3 示例 4.操作元素 4.1 获取/修改元素内容 4.2 获取/修改元素属性 4.3 获取/修改表单元素属性 4.3.1 value&#xf…
阅读更多...
MES系统有哪些厂家？万界星空科技是您不容错过的选择

MES系统有哪些厂家？万界星空科技是您不容错过的选择

为什么要选择MES系统&#xff1f; MES系统可以帮助企业实现生产计划的有效执行和优化&#xff0c;提高生产效率和质量&#xff0c;并实现产能的最大化利用。它可以实现对生产过程的全面监控和调度&#xff0c;使企业能够及时获得生产数据和指标&#xff0c;并做出相应的决策。…
阅读更多...
Linux操作系统基础（十三）：Linux安装、卸载MySQL

Linux操作系统基础（十三）：Linux安装、卸载MySQL

文章目录 Linux安装、卸载MySQL 一、卸载系统自带的mariadb-lib 二、上传安装包并解压 三、按顺序安装 错误1: 错误2: 错误3: 错误4: 四、初始化数据库 五、目录授权&#xff0c;否则启动失败 六、启动msyql服务 七、查看msyql服务的状态 八、在/var/log/mysqld.l…
阅读更多...
一周学会Django5 Python Web开发-项目配置settings.py文件-基本配置

一周学会Django5 Python Web开发-项目配置settings.py文件-基本配置

锋哥原创的Python Web开发 Django5视频教程&#xff1a; 2024版 Django5 Python web开发 视频教程(无废话版) 玩命更新中~_哔哩哔哩_bilibili2024版 Django5 Python web开发 视频教程(无废话版) 玩命更新中~共计17条视频&#xff0c;包括&#xff1a;2024版 Django5 Python we…
阅读更多...
安卓价值2-Macrodroid在其它app下执行两步就停

安卓价值2-Macrodroid在其它app下执行两步就停

Macrodroid 是一款适用于 Android 平台的自动化应用程序。它允许用户创建个性化的自动化工作流程，以简化日常任务并增强手机的功能。 但使用下来会发现一些奇怪的问题，比如在其它app处于前台状态下它执行了两步任务就停止了，但切换回macrodroid就又继续执行了，这就像是程序…
阅读更多...
Selenium实战教程系列（二）---元素定位

Selenium实战教程系列（二）---元素定位

Selenium webdriver能够模拟人对浏览器进行操作的前提是界面元素的定位。元素的定位可以说是Selenium自动化脚本的基础。这一小节笔者将介绍如何在selenium中进行元素的定位。 定位元素的方法 Selenium中提供了以下定位元素的方法&#xff1a; 首先看一个HTML文件 test_page.…
阅读更多...
Acwing 周赛142 解题报告 | 珂学家 | BFS集合

Acwing 周赛142 解题报告 | 珂学家 | BFS集合

前言 整体评价 VP了这场比赛&#xff0c;感觉T2挺有意思的&#xff0c;超级容易错&#xff0c;T3到时中规中矩&#xff0c;算Middle更合适。 A. 倒序排列 思路: 模拟 n int(input())l [i for i in range(n, 0, -1)]print (*l, sep )B. 最有价值字符串 思路: 思维 这题难…
阅读更多...
PHP开发日志 ━━ 深入理解三元操作与一般条件语句的不同

PHP开发日志 ━━ 深入理解三元操作与一般条件语句的不同

概况 三元运算符的功能与“if…else”流程语句一致。 在一般情况下&#xff0c;三元操作替换if条件语句可以精简代码&#xff0c;并且更为直观&#xff0c;但是在下面的情况中使用三元操作将会返回警告。 借图&#xff1a; 案例 比如原代码&#xff1a; class classA{publ…
阅读更多...
Python算法探索：从经典到现代

Python算法探索：从经典到现代

引言 Python&#xff0c;作为一种功能强大的编程语言&#xff0c;一直是算法实现的首选工具。从经典的排序和查找算法到现代的机器学习和深度学习算法&#xff0c;Python都展现出了其强大的实力。接下来&#xff0c;我们将一起探索Python算法的经典与现代。 一、经典算法&#…
阅读更多...
DolphinScheduler安装与配置

DolphinScheduler安装与配置

DolphinScheduler概述 Apache DolphinScheduler是一个分布式、易扩展的可视化DAG工作流任务调度平台。致力于解决数据处理流程中错综复杂的依赖关系&#xff0c;使调度系统在数据处理流程中开箱即用。 DolphinScheduler的主要角色如下&#xff1a; MasterServer采用分布式无…
阅读更多...
HarmonyOS鸿蒙学习基础篇 - 自定义组件（一）

HarmonyOS鸿蒙学习基础篇 - 自定义组件（一）

前言 在ArkUI中&#xff0c;UI显示的内容均为组件&#xff0c;由框架直接提供的称为系统组件&#xff0c;由开发者定义的称为自定义组件。在进行 UI 界面开发时&#xff0c;通常不是简单的将系统组件进行组合使用&#xff0c;而是需要考虑代码可复用性、业务逻辑与UI分离&#…
阅读更多...
一起玩儿Proteus仿真（C51）——06. 红绿灯仿真（二）

一起玩儿Proteus仿真（C51）——06. 红绿灯仿真（二）

摘要&#xff1a;本文介绍如何仿真红绿灯 今天来看一下红绿灯仿真程序的具体实现方法。先来看一下整个程序的原理图。 在这个红绿灯仿真实验中&#xff0c;每个路口需要控制的设备是2位数码管显示倒计时以及红黄绿灯的亮灭。先来看一下数码管的连接方法。 数码管的8根LED显示…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023