Kafka 位移提交
Kafka 位移提交
- 自动提交
- 手动提交
Consumer 的消费位移 : 记录 Consumer 下一条消息的消费位移
- 如 : Consumer 已消费 5 条消息 (位移: 0 - 4) , 此时 Consumer 位移 = 5 : 指向下一条消息的位移
提交位移 (Committing Offsets) : Consumer 向 Kafka 汇报位移数据
- Consumer 能同时消费多个分区的数据,Consumer 要维护每个分区提交各自的位移数据
- 当 Consumer 重启后,能从之前位移继续消费,避免重新消费整个消息
Consumer API 的提交位移的方法 :
- 从用户分 : 自动提交 , 手动提交
- 从 Consumer 分 : 同步提交 , 异步提交
- 自动提交 : Consumer 在后台提交位移,用户无需操作
- 手动提交 : 用户提交位移,Consumer 不管
| 提交位移 | 自动提交 | 配置 | enable.auto.commit = true |
|---|---|---|---|
| 手动提交 | 同步提交 | KafkaConsumer.commitSync | |
| 异步提交 | KafkaConsumer.commitAsync | ||
| 细化位移提交 | commitSync(Map<TopicPartition, OffsetAndMetadata>) | ||
commitAsync(Map<TopicPartition, OffsetAndMetadata>) |
自动提交
Consumer 参数 :
enable.auto.commit = true: 自动提交位移auto.commit.interval.ms(默认值是 5 秒) : Kafka 每 5 秒自动提交一次位移
自动提交位移 :
- 可能出现重复消费
- 例子:Consumer 每 5 秒自动提交一次位移。提交位移 3 秒后出现 Rebalance。在 Rebalance 后,所有 Consumer 从上一次提交的位移处继续消费,但该位移已经是 3 秒前的位移数据,在 Rebalance 发生前 3 秒消费的所有数据都会重新消费
设置自动提交位移 :
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "2000");
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("foo", "bar"));while (true) {ConsumerRecords<String, String> records = consumer.poll(100);for (ConsumerRecord<String, String> record : records) {System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());}
}
手动提交
enable.auto.commit = false : 手动提交位移
手动提交位移 :
- 好处 : 更灵活,能把控位移提交的时机和频率
- 缺点 : 用 commitSync() 时,Consumer 处于阻塞状态,直到 Broker 返回提交结果,影响整个应用程序的 TPS
commitSync() :
while (true) {// 返回最新位移。一直等位移提交后才返回 (同步操作)ConsumerRecords<String, String> records =consumer.poll(Duration.ofSeconds(1));process(records); // 处理消息try {consumer.commitSync();} catch (CommitFailedException e) {handle(e); // 处理提交失败异常}
}
commitAsync() :
- 异步操作,会立即返回,不会阻塞,不影响 Consumer 的 TPS
- 用回调函数 (callback) 实现提交后的逻辑,如 : 记录日志或处理异常
- 无法自动失败重试
while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));process(records); // 处理消息consumer.commitAsync((offsets, exception) -> {if (exception != null)handle(exception);});
}
异步无阻塞式 :
- 用 commitSync 自动重试避免瞬时错误,如 : 网络的瞬时抖动,Broker 端 GC
- 异步处理,不影响 TPS
// 实现异步无阻塞式的位移管理,保证 Consumer 位移的正确性
try {while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));process(records); // 处理消息commitAysnc(); // 使用异步提交规避阻塞}
} catch (Exception e) {handle(e); // 处理异常
} finally {try {consumer.commitSync(); // 最后一次提交使用同步阻塞式提交} finally {consumer.close();}
}
更精细的位移管理 :
commitSync(Map<TopicPartition, OffsetAndMetadata>)commitAsync(Map<TopicPartition, OffsetAndMetadata>)- 参数 : Map 对象 : 键 = TopicPartition (消费的分区),值 = OffsetAndMetadata 对象 (位移数据)
// 创建 Map 对象,保存 Consumer 消费要提交的分区位移
private Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
int count = 0;
//...
while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));for (ConsumerRecord<String, String> record: records) {process(record); // 处理消息// 构造要提交的位移值offsets.put(new TopicPartition(record.topic(), record.partition()),new OffsetAndMetadata(record.offset() + 1);// 每 100 条消息提交一次位移if(count % 100 == 0){consumer.commitAsync(offsets, null); // 回调处理逻辑是 null}count++;}
}
相关文章:
Kafka 位移提交
Kafka 位移提交自动提交手动提交Consumer 的消费位移 : 记录 Consumer 下一条消息的消费位移 如 : Consumer 已消费 5 条消息 (位移: 0 - 4) , 此时 Consumer 位移 5 : 指向下一条消息的位移 提交位移 (Committing Offsets) : Consumer 向 Kafka 汇报位移数据 Consumer 能同…...
kubernetes--监控容器运行时:Falco
目录 Falco介绍 Falco架构 Falco的安装 告警规则示列 威胁场景测试: 监控容器创建的不可信任进程(自定义规则) Falco支持五种输出告警方式falco.yaml: Falco告警集中化展示: Falco介绍 Falco是一个Linux安全工具…...
HTTP协议详解(上)
目录 前言: 认识URL HTTP协议方法 通过Fiddler抓包 GET和POST之间典型区别 header详解 HTTP响应状态码 常见状态码解释 状态码分类 HTTP协议报文格式 小结: 前言: HTTP协议属于应用层协议,称为超文本传输协议ÿ…...
java性能-原生内存-内存分析
原生内存最佳实践 内存占用 jVM使用的原生内存和堆内存总和就是一个应用程序的总内存——操作系统角度 jvm启动时候加载的类路径下的jar文件相关的内存和系统其他进程共享资源的可能 测量内存占用 线程是个例外——每当创建一个线程操作系统都会分配一些原生内存存储线程栈…...
c++类与对象
🐶博主主页:ᰔᩚ. 一怀明月ꦿ ❤️🔥专栏系列:线性代数,C初学者入门训练,题解C,C的使用文章 🔥座右铭:“不要等到什么都没有了,才下定决心去做” …...
Java并发编程与API详解
文章目录前言操作系统——进程和线程进程进程组成进程状态进程控制进程创建进程终止进程阻塞和唤醒进程通信线程线程组成线程状态线程控制线程的实现方式用户线程内核线程混合方式CPU调度调度的层次调度的实现调度器调度的时机、切换与过程进程调度的方式闲逛进程两种线程的调度…...
【冲刺蓝桥杯的最后30天】day5
大家好😃,我是想要慢慢变得优秀的向阳🌞同学👨💻,断更了整整一年,又开始恢复CSDN更新,从今天开始更新备战蓝桥30天系列,一共30天,如果对你有帮助或者正在备…...
大厂与小厂招人的区别,看完多少有点不敢相信
前两天在头条发了一条招人的感慨,关于大厂招人和小公司招人的区别。 大厂:有影响力,有钱,能够吸引了大量的应聘者。因此,也就有了筛选的资格,比如必须985名校毕业,必须35岁以下,不能…...
前端ES5对象特性
ES5对象特性 对象和函数的原型 JS中每一个对象都有一个特殊的内置属性,这个特殊的对象可以指向其他的对象 我们通过引用对象的属性key来获取一个value时,它会触发 Get 的操作首先检查该对象是否有对应的属性,如果有的话就使用对象内的如果…...
Linux入门介绍及Linux文件与目录结构
前言 本文小新为大家带来 Linux 入门介绍及Linux 文件与目录结构 相关知识,具体内容包括Linux入门介绍(包括:Linux概述,Linux与Windows区别,CentOS 下载地址),Linux文件与目录结构等进行详尽介绍…...
超赞,用python实现流媒体服务器功能,寥寥几句搞定。
步骤: 要使用Python将实时摄像机传送流写入H5页面,可以使用以下步骤。 1、安装必要的软件包。您需要安装OpenCV和Flask以及gunicorn 与 gevent 。您可以通过在终端中运行以下命令来执行此操作。 pip install opencv-python pip install Flask pip ins…...
冥想第七百二十一天
1.3.3周五,又是周五了。今天又运动了5公里,很舒服轻松。 2.还是往常的生活,休息的也很好,开春后跑的一直很好。 3.早上30分钟健康操。中午转了圈, 给大哥说下周去上海。 4.感谢父母,感谢朋友,感…...
06-Oracle表空间与用户管理
本讲主要内容: 1.表空间管理:表空间的作用,创建,修改,删除及管理; 2.用户管理:创建用户,修改用户,删除用户,修改密码,解锁; 3.用户…...
Mysql 索引特点
承接上文Mysql Server原理简介聚簇索引、二级索引、联合索引分别具备什么样的特点?聚簇索引数据跟索引放在一起的叫聚簇索引;数据和索引分开存储的叫非聚簇索引;innodb存储引擎,数据和文件都放在ibd文件中,实际的数据是…...
读书笔记-终身学习
前言人需要终身成长,也需要终身学习,以下是记录个人读书学习的笔记总结,希望能给大家一点借鉴,仅供参考。笔记1、《匠人精神》秋山利辉是日本神奈川县横滨市都筑区“秋山木工”的经营者,从事订制家具制作业务。是一家小…...
了解栈Stack一篇文章就够了
什么是栈栈是一种特殊的线性表,只允许一端进行数据的插入和删除,即先进后出原则。类似于弹夹先装进去的子弹后面出,后放入的子弹先出。栈的底层原理栈是一种线性结构,所以既能使用数组实现,也能使用链表实现࿰…...
CNStack 助推龙源电力扛起“双碳”大旗
作者:CNStack 容器平台、龙源电力:张悦超 、党旗 龙源电力容器云项目背景 龙源电力集团是世界第一大风电运营商, 随着国家西部大开发战略推进,龙源电力已经把风力发电场铺设到全国各地,甚至是交通极不便利的偏远地区&…...
ruoyi-vue-plus1(控制台相关的输出日志)(p6spy插件)(jackson全局配置)(StopWatch)
Jackson配置在启动项目时,我们发现日志打印出这样几行字,初始化了jacdson配置,我们去查看一下来源找。我们找到了一个全局序列化配置类,其中重写了BigNumberSerializer.INSTANCE进去查看发现了这里对于部分范围的数字进行了转为为…...
【Mybatis】| 如何创建MyBatis的工具类
目录🌟更多专栏请点击👇一、前言二、实现过程1. 创建一个ThreadLocal对象2. 初始化SqlSessionFactory3. 获取并存储sqlSession对象4. 关闭sqlSession对象三、 总代码🌟更多专栏请点击👇 专栏名字🔥Elasticsearch专栏e…...
【Java】DT怎么写?
几个重要的注解 怎么用mockito写单元测试? package Biz;import Client.FileIOClient; import Req.FileRequest; import Res.FileResponse; import org.junit.Assert; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.InjectMocks;…...
8k长序列建模,蛋白质语言模型Prot42仅利用目标蛋白序列即可生成高亲和力结合剂
蛋白质结合剂(如抗体、抑制肽)在疾病诊断、成像分析及靶向药物递送等关键场景中发挥着不可替代的作用。传统上,高特异性蛋白质结合剂的开发高度依赖噬菌体展示、定向进化等实验技术,但这类方法普遍面临资源消耗巨大、研发周期冗长…...
苍穹外卖--缓存菜品
1.问题说明 用户端小程序展示的菜品数据都是通过查询数据库获得,如果用户端访问量比较大,数据库访问压力随之增大 2.实现思路 通过Redis来缓存菜品数据,减少数据库查询操作。 缓存逻辑分析: ①每个分类下的菜品保持一份缓存数据…...
CocosCreator 之 JavaScript/TypeScript和Java的相互交互
引擎版本: 3.8.1 语言: JavaScript/TypeScript、C、Java 环境:Window 参考:Java原生反射机制 您好,我是鹤九日! 回顾 在上篇文章中:CocosCreator Android项目接入UnityAds 广告SDK。 我们简单讲…...
Rust 异步编程
Rust 异步编程 引言 Rust 是一种系统编程语言,以其高性能、安全性以及零成本抽象而著称。在多核处理器成为主流的今天,异步编程成为了一种提高应用性能、优化资源利用的有效手段。本文将深入探讨 Rust 异步编程的核心概念、常用库以及最佳实践。 异步编程基础 什么是异步…...
实现弹窗随键盘上移居中
实现弹窗随键盘上移的核心思路 在Android中,可以通过监听键盘的显示和隐藏事件,动态调整弹窗的位置。关键点在于获取键盘高度,并计算剩余屏幕空间以重新定位弹窗。 // 在Activity或Fragment中设置键盘监听 val rootView findViewById<V…...
听写流程自动化实践,轻量级教育辅助
随着智能教育工具的发展,越来越多的传统学习方式正在被数字化、自动化所优化。听写作为语文、英语等学科中重要的基础训练形式,也迎来了更高效的解决方案。 这是一款轻量但功能强大的听写辅助工具。它是基于本地词库与可选在线语音引擎构建,…...
代码规范和架构【立芯理论一】(2025.06.08)
1、代码规范的目标 代码简洁精炼、美观,可持续性好高效率高复用,可移植性好高内聚,低耦合没有冗余规范性,代码有规可循,可以看出自己当时的思考过程特殊排版,特殊语法,特殊指令,必须…...
HybridVLA——让单一LLM同时具备扩散和自回归动作预测能力:训练时既扩散也回归,但推理时则扩散
前言 如上一篇文章《dexcap升级版之DexWild》中的前言部分所说,在叠衣服的过程中,我会带着团队对比各种模型、方法、策略,毕竟针对各个场景始终寻找更优的解决方案,是我个人和我司「七月在线」的职责之一 且个人认为,…...
微服务通信安全:深入解析mTLS的原理与实践
🔥「炎码工坊」技术弹药已装填! 点击关注 → 解锁工业级干货【工具实测|项目避坑|源码燃烧指南】 一、引言:微服务时代的通信安全挑战 随着云原生和微服务架构的普及,服务间的通信安全成为系统设计的核心议题。传统的单体架构中&…...
OpenGL-什么是软OpenGL/软渲染/软光栅?
软OpenGL(Software OpenGL)或者软渲染指完全通过CPU模拟实现的OpenGL渲染方式(包括几何处理、光栅化、着色等),不依赖GPU硬件加速。这种模式通常性能较低,但兼容性极强,常用于不支持硬件加速…...
