Kafka 使用java api从指定位移消费 （从开头消费/从结尾消费）

article/2025/8/22 9:00:41

一、auto.offset.reset值详解

在 Kafka 中，每当消费者组内的消费者查找不到所记录的消费位移或发生位移越界时，就会根据消费者客户端参数 auto.offset.reset 的配置来决定从何处开始进行消费，这个参数的默认值为 “latest” 。

auto.offset.reset 的值可以为 earliest、latest 和 none 。关于 earliest 和 latest 的解释，官方描述的太简单，各含义在真实情况如下所示：

  • earliest ：当各分区下存在已提交的 offset 时，从提交的 offset 开始消费；无提交的 offset 时，从头开始消费。
  • latest ：当各分区下存在已提交的 offset 时，从提交的 offset 开始消费；无提交的 offset 时，消费该分区下新产生的数据。
  • none ：topic 各分区都存在已提交的 offset 时，从 offset 后开始消费；只要有一个分区不存在已提交的offset，则抛出异常。

二、seek()方法

到目前为止，我们知道消息的拉取是根据 poll() 方法中的逻辑来处理的，这个 poll() 方法中的逻辑对于普通的开发人员而言是一个黑盒，无法精确地掌控其消费的具体位置。Kafka 提供的 auto.offset.reset 参数也只能在找不到消费位移或位移越界的情况下粗粒度地从开头或末尾开始消费。有的时候，我们需要一种更细粒度的掌控，可以让我们从指定的位移处开始拉取消息，而 KafkaConsumer 中的 seek() 方法正好提供了这个功能，让我们得以追前消费或回溯消费。seek() 方法的具体定义如下：

public void seek(TopicPartition partition, long offset)

seek() 方法中的参数 partition 表示分区，而 offset 参数用来指定从分区的哪个位置开始消费。seek() 方法只能重置消费者分配到的分区的消费位置，而分区的分配是在 poll() 方法的调用过程中实现的，也就是说，在执行 seek() 方法之前需要先执行一次 poll() 方法，等到分配到分区之后才可以重置消费位置。

如果对未分配的分区执行 seek() 方法，那么会报出 IllegalStateException 的异常。类似在调用 subscribe() 方法之后直接调用 seek() 方法，如下所示：

consumer.subscribe(Arrays.asList(TOPIC));
consumer.seek(new TopicPartition(TOPIC, 0), 80);

会报下述错误：
在这里插入图片描述

三、指定offset开始消费

接下来的代码示例讲述了消费各分区 offset 为 80（包括80）之后的消息：

Properties props = initConfig();
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList(TOPIC));Set<TopicPartition> assignment = new HashSet<>();
// 在poll()方法内部执行分区分配逻辑，该循环确保分区已被分配。
// 当分区消息为0时进入此循环，如果不为0，则说明已经成功分配到了分区。
while (assignment.size() == 0) {consumer.poll(100);// assignment()方法是用来获取消费者所分配到的分区消息的// assignment的值为：topic-demo-3, topic-demo-0, topic-demo-2, topic-demo-1assignment = consumer.assignment();
}
System.out.println(assignment);for (TopicPartition tp : assignment) {int offset = 80;System.out.println("分区 " + tp + " 从 " + offset + " 开始消费");consumer.seek(tp, offset);
}while (true) {ConsumerRecords<String, String> records = consumer.poll(1000);// 消费记录for (ConsumerRecord<String, String> record : records) {System.out.println(record.offset() + ":" + record.value() + ":" + record.partition());}
}

**注意：**假如某分区的前 100 条数据由于过期，导致被删除，那么此时如果使用 seek() 方法指定 offset 为 0 进行消费的话，是消费不到数据的。因为前 100 条数据已被删除，所以只能从 offset 为 100 ，来进行消费。

四、从分区开头或末尾开始消费

如果消费者组内的消费者在启动的时候能够找到消费位移，除非发生位移越界，否则 auto.offset.reset 参数不会奏效。此时如果想指定从开头或末尾开始消费，也需要 seek() 方法来实现。

如果按照第三节指定位移消费的话，就需要先获取每个分区的开头或末尾的 offset 了。可以使用 beginningOffsets() 和 endOffsets() 方法。

  • public Map beginningOffsets(Collection partitions)
  • public Map beginningOffsets(Collection partitions, long timeout)
  • public Map endOffsets(Collection partitions)
  • public Map endOffsets(Collection partitions, long timeout)

其中 partitions 参数表示分区集合，而 timeout 参数用来设置等待获取的超时时间。如果没有指定 timeout 的值，那么 timeout 的值由客户端参数 request.timeout.ms 来设置，默认为 30000 。

接下来通过示例展示如何从分区开头或末尾开始消费：

Set<TopicPartition> assignment = new HashSet<>();
// 在poll()方法内部执行分区分配逻辑，该循环确保分区已被分配。
// 当分区消息为0时进入此循环，如果不为0，则说明已经成功分配到了分区。
while (assignment.size() == 0) {consumer.poll(100);// assignment()方法是用来获取消费者所分配到的分区消息的// assignment的值为：topic-demo-3, topic-demo-0, topic-demo-2, topic-demo-1assignment = consumer.assignment();
}// 指定分区从头消费
Map<TopicPartition, Long> beginOffsets = consumer.beginningOffsets(assignment);
for (TopicPartition tp : assignment) {Long offset = beginOffsets.get(tp);System.out.println("分区 " + tp + " 从 " + offset + " 开始消费");consumer.seek(tp, offset);
}// 指定分区从末尾消费
Map<TopicPartition, Long> endOffsets = consumer.endOffsets(assignment);
for (TopicPartition tp : assignment) {Long offset = endOffsets.get(tp);System.out.println("分区 " + tp + " 从 " + offset + " 开始消费");consumer.seek(tp, offset);
}// 再次执行poll()方法，消费拉取到的数据。
// ...(省略)

值得一说的是：

  • 指定分区从头消费时，需要了解：一个分区的起始位置是 0 ，但并不代表每时每刻都为 0 ，因为日志清理的动作会清理旧的数据，所以分区的起始位置会自然而然地增加。
  • 指定分区从末尾消费，需要了解：endOffsets() 方法获取的是将要写入最新消息的位置。

其实，KafkaConsumer 中直接提供了 seekToBeginning() 和 seekToEnd() 方法来实现上述功能。具体定义如下：

  • public void seekToBeginning(Collection partitions)
  • public void seekToEnd(Collection partitions)

例如使用

consumer.seekToBeginning(assignment);

直接可以代替

Map<TopicPartition, Long> beginOffsets = consumer.beginningOffsets(assignment);
for (TopicPartition tp : assignment) {Long offset = beginOffsets.get(tp);System.out.println("分区 " + tp + " 从 " + offset + " 开始消费");consumer.seek(tp, offset);
}

五、根据时间戳消费

有时候我并不知道特定的消费位置，却知道一个相关的时间点。比如我想要消费某个时间点之后的消息，这个需求更符合正常的思维逻辑。这时，我们可以用 offsetsForTimes() 方法，来获得符合筛选条件的 offset ，然后再结合 seek() 方法来消费指定数据。offsetsForTimes() 方法如下所示：

  • public Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch)

offsetsForTimes() 方法的参数 timestampsToSearch 是一个 Map 类型，其中 key 为待查询的分区，value 为待查询的时间戳，该方法会返回时间戳大于等于查询时间的第一条消息对应的 offset 和 timestamp 。

接下来就以消费当前时间前一天之后的消息为例，代码片段如下所示：

Set<TopicPartition> assignment = new HashSet<>();
// 在poll()方法内部执行分区分配逻辑，该循环确保分区已被分配。
// 当分区消息为0时进入此循环，如果不为0，则说明已经成功分配到了分区。
while (assignment.size() == 0) {consumer.poll(100);// assignment()方法是用来获取消费者所分配到的分区消息的// assignment的值为：topic-demo-3, topic-demo-0, topic-demo-2, topic-demo-1assignment = consumer.assignment();
}Map<TopicPartition, Long> timestampToSearch = new HashMap<>();
for (TopicPartition tp : assignment) {// 设置查询分区时间戳的条件：获取当前时间前一天之后的消息timestampToSearch.put(tp, System.currentTimeMillis() - 24 * 3600 * 1000);
}// timestampToSearch的值为{topic-demo-0=1563709541899, topic-demo-2=1563709541899, topic-demo-1=1563709541899}
Map<TopicPartition, OffsetAndTimestamp> offsets = consumer.offsetsForTimes(timestampToSearch);for(TopicPartition tp: assignment){// 获取该分区的offset以及timestampOffsetAndTimestamp offsetAndTimestamp = offsets.get(tp);// 如果offsetAndTimestamp不为null，则证明当前分区有符合时间戳条件的消息if (offsetAndTimestamp != null) {consumer.seek(tp, offsetAndTimestamp.offset());}
}while (true) {ConsumerRecords<String, String> records = consumer.poll(1000);System.out.println("##############################");System.out.println(records.count());// 消费记录for (ConsumerRecord<String, String> record : records) {System.out.println(record.offset() + ":" + record.value() + ":" + record.partition() + ":" + record.timestamp());}
}

六、总结
本文内容主要讲解了消费者如何指定位移消费，主要从以下几方面入手：

  • 讲解了 auto.offset.reset 参数值的含义。
  • 如何使用 seek() 方法指定 offset 消费。
  • 接着又介绍了如何从分区开头或末尾消费消息：beginningOffsets()、endOffsets()、seekToBeginning、seekToEnd() 方法。
  • 最后又介绍了如何根据时间戳来消费指定消息，更加务实一些。

即使消息已被提交，但我们依然可以使用 seek() 方法来消费符合一些条件的消息，这样为消息的消费提供了很大的灵活性。


http://chatgpt.dhexx.cn/article/TfRrugVG.shtml

相关文章

动态SQL之 where 标签

动态SQL之 where 标签

动态SQL之 where 标签 where和if一般结合使用&#xff1a; 1.若where标签中的 if 条件都不满足&#xff0c;则where标签没有任何功能&#xff0c;即不会添加where关键字 2.若where标签中的 if 条件满足&#xff0c;则where标签会自动添加where关键字&#xff0c;并将条件最前…
阅读更多...
mybatis-动态sql

mybatis-动态sql

文章目录 1. 动态sql简述2. 动态sql示例 2.1 if2.2 choose2.3 foreach2.4 sql 及 include2.5 sql中的特殊字符3. 后台分页实现4. 数据版本号处理并发问题 1. 动态sql简述 mybatis的动态sql语句是基于OGNL表达式的。可以方便的在sql语句中实现某些逻辑. 总体说来mybatis动态SQL…
阅读更多...
mysql动态sql拼接_动态SQL（拼接）

mysql动态sql拼接_动态SQL（拼接）

Q1：什么是动态SQL呢？ A1：首先是SQL语句，是根据条件来拼接SQL Q2：为什么要用动态SQL？ A2：因为在条件WHERE中出现OR会导致不能使用索引，从而使效率差别巨大。 例如：如图1、2, 图(1) 图(2) Q3：怎么样使用动态SQL？ A3： 存储过程Proc_Test是没有采用拼接的：CREATE PROC…
阅读更多...
Mybatis学习之动态Sql

Mybatis学习之动态Sql

目录 1. 什么是动态Sql 2. 动态Sql需要学习什么 3. 动态Sql之《if》 4. 动态Sql之《where》 5. 动态Sql之《foreach》 6. 动态Sql之《sql》 7. PageHelper分页插件的使用 1. 什么是动态Sql 答案&#xff1a;动态Sql指的是&#xff0c;Sql语句是变化的&#xff0c;不是固…
阅读更多...
Mybatis 动态SQL

Mybatis 动态SQL

Mybatis 动态SQL 一 .动态SQL 数组 array 使用foreach 标签 <!-- mybatis的集合操作知识点: 如果遇到集合参数传递,需要将集合遍历标签: foreach 循环遍历集合标签属性说明:1.collection 表示遍历的集合类型1.1 数组 关键字 array1.2 List集合 关键字 list1.3 Map集…
阅读更多...
Mybatis动态SQL解析

Mybatis动态SQL解析

文章目录 1 为什么需要动态SQL&#xff1f;2 动态标签有哪些?3 举例说明ifchoose (when, otherwise)trim (where, set)foreach 1 为什么需要动态SQL&#xff1f; 看一段Oracle存储过程代码&#xff1a; 由于前台传入的查询参数不同&#xff0c;所以写了很多的if else&#x…
阅读更多...
Java MyBatis动态SQL

Java MyBatis动态SQL

&#x1f9ed;MyBatis学习 &#x1f389; 内容回顾 Java MyBatis的介绍及其执行原理 Java MyBatis配置详解 Java Mybatis中使用Junit进行测试_程序员必备 Java MyBatis的使用 &#x1f4e2;今天我们进行 Java MyBatis动态SQL 的学习&#xff0c;感谢你的阅读&#xff0c;内容若…
阅读更多...
mysql动态SQL用法

mysql动态SQL用法

顾名思义“动态”SQL就是不固定的SQL&#xff0c;根据不同的条件把SQL语句进行拼接&#xff0c;来实现对数据库更加精准的操作。可以通过配置文件或者注解的形式实现&#xff0c;多用于多条件联查。 xml版&#xff08;配置文件&#xff09;&#xff1a; xml版的动态SQL 接…
阅读更多...
动态SQL标签

动态SQL标签

所谓的动态SQL&#xff0c;本职还是SQL语句&#xff0c;只是可以在SQL层面&#xff0c;去执行一个逻辑代码 动态SQL就是在拼接SQL语句&#xff0c;我们只要保证SQL的正确性&#xff0c;按照SQL的格式&#xff0c;去排列组合。 建议&#xff1a; 先在MySQL中写出完整的SQL&am…
阅读更多...
MyBatis动态SQL

MyBatis动态SQL

文章目录 前言一、\<if\>标签二、\<where\>标签三、\<trim\>标签四、\<set\>标签五、\<foreach\>标签五、\<sql\>标签 与 \<include\>标签 前言 动态sql是Mybatis的强大功能特性之一&#xff0c;能够完成不同条件下的sql拼接 以上…
阅读更多...
动态SQL

动态SQL

动态SQL 在项目开发中&#xff0c;动态SQL可以解决很多不确定因素导致的SQL语句不同的问题。动态SQL可以简单高效的进行编码。在接下来的案例中进行认识和学习动态SQL。 动态SQL只是在原有的SQL语句中进行细微修改。案例贴合实际&#xff0c;编码简单易懂 文章目录 动态SQL一、…
阅读更多...
动态 SQL

动态 SQL

文章目录 一、学习目的二、动态 SQL 中的元素三、条件查询操作四、更新操作五、复杂查询操作1.foreach 元素中的属性2.foreach 元素迭代数组3.foreach 元素迭代 List4.foreach 元素迭代 Map 一、学习目的 在实际项目的开发中&#xff0c;开发人员在使用 JDBC 或其他持久层框架…
阅读更多...
第3章 动态SQL

第3章 动态SQL

目录/Contents 第3章 动态SQL学习目标掌握MyBatis中动态SQL元素的使用掌握MyBatis的条件查询操作掌握MyBatis的更新操作掌握MyBatis的复杂查询操作 学习内容1 动态SQL中的元素1.1 使用动态SQL的好处1.2 动态SQL常用元素 2 条件查询操作2.1 \<if>元素2.2 \<choose>、…
阅读更多...
Mybatis的特性详解——动态SQL

Mybatis的特性详解——动态SQL

Mybatis的特性详解——动态SQL 前言一、动态sql的元素1.MyBatis if标签&#xff1a;条件判断2.MyBatis choose、when和otherwise标签3.MyBatis where标签4.MyBatis set标签5.MyBatis foreach标签6.MyBatis bind标签7.MyBatis trim标签仰天大笑出门去&#xff0c;我辈岂是蓬蒿人…
阅读更多...
Java编程基础八股文（背诵版）

Java编程基础八股文（背诵版）

Java语言具有哪些特点&#xff1f; Java为纯面向对象的语言。它能够直接反应现实生活中的对象。 具有平台无关性。java利用Java虚拟机运行字节码&#xff0c;无论是在Windows、Linux还是MacOS等其它平台对Java程序进行编译&#xff0c;编译后的程序可在其它平台运行。 Java为…
阅读更多...
Java编程基础篇

Java编程基础篇

目录 计算机、程序和Java概述计算机组成 基本程序设计标识符数值数据类型格式化控制台输出输入输出重定向输入重定向输出重定向输入输出重定向方法重载变量的范围 数组声明数组变量创建数组初始化数组可边长参数二维数组静态变量、常量和方法静态变量 限定字符串StringBuilder和…
阅读更多...
Java 基础编程入门

Java 基础编程入门

一、什么是Java Java是一种编程语言二、应用场景 非windows平台下互联网环境的开发首选三、Java语言特点 1、应用面广 2、简单易学(相对于C/C) 3、面向对象(更符合我们看待事物的特点) 4、跨平台(一处编译&#xff0c;处处运行) 5、多线程(提升系统性能) 单线…
阅读更多...
JAVA的编程基础（上）

JAVA的编程基础（上）

1. JAVA的基本语法 1.1 注释 单行注释&#xff1a;// 多行注释&#xff1a;/* */ 文档注释&#xff1a;/** */ 注意&#xff1a;单行注释可以嵌套&#xff0c;多行注释和文档注释不可以嵌套 1.2 关键字 含义&#xff1a;在编程语言中&#xff0c;已经被赋予一些特殊含义…
阅读更多...
Java编程基础（1）

Java编程基础（1）

1.Java基本语法 &#xff08;1&#xff09;Java程序的基本格式&#xff1a; 修饰符 class 类名 { 程序代码 } 如果Java程序要运行&#xff0c;则必须要有main方法&#xff0c;main方法是程序运行的入口 格式&#xff1a; 修饰符 class 类名{ public static void main&…
阅读更多...
Java编程基础之网络编程

Java编程基础之网络编程

网络编程 文章目录 网络编程概述C/S和B/S网络通信协议网络编程要素 IP类UDP协议数据传输发送端接收端udp数据传输丢失问题 TCP概述Socket客户端ServerSocket服务端Socket**服务器代码**客户端代码 文件上传实现客户端服务端多线程版 概述 C/S和B/S 网络通信协议 协议:protoco…
阅读更多...
推荐文章

Copyright @ 2022~2023