java代码实现kafka消费端consumer的from-beginning功能

只需要在代码中加入

props.put("auto.offset.reset", "earliest");
props.put("group.id", UUID.randomUUID().toString());

完整例子

//1、准备配置文件Properties props = new Properties();props.put("bootstrap.servers", "hadoop1:9092");props.put("acks", "all");props.put("retries", 0);props.put("batch.size", 16384);props.put("linger.ms", 1);props.put("buffer.memory", 33554432);props.put("group.id", "test");props.put("enable.auto.commit", "true");props.put("auto.commit.interval.ms", "1000");props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("auto.offset.reset", "earliest");props.put("group.id", UUID.randomUUID().toString());// 2、创建KafkaConsumerKafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(props);// 3、订阅数据，这里的topic可以是多个kafkaConsumer.subscribe(Arrays.asList("yun03"));// 4、获取数据while (true) {ConsumerRecords<String, String> records = kafkaConsumer.poll(100);for (ConsumerRecord<String, String> record : records) {System.out.printf("topic = %s,offset = %d, key = %s, value = %s%n",record.topic(), record.offset(), record.key(), record.value());}}

java代码实现kafka消费端consumer的from-beginning功能相关推荐

解决Kafka消费端错误：o.s.kafka.listener.LoggingErrorHandler : Error while processing: null
简单记录下使用kafka遇到的问题,万一也会有小伙伴遇到了呢. 程序端使用springboot服务消费kafka,某天出现消息大量堆积,经过定位到错误日志如下: 2020-05-12 10:22:36 ...
Kafka设计解析（十三）Kafka消费组(consumer group)
转载自 huxihx,原文链接 Kafka消费组(consumer group) 一直以来都想写一点关于kafka consumer的东西,特别是关于新版consumer的中文资料很少.最近Kafka ...
kafka 消费端 api_在消费者的眼中：您真的需要为您的API提供客户端库吗？
kafka 消费端 api RESTful Web服务和API的优点在于,任何使用HTTP协议的使用者都可以理解和使用它. 但是,同样的难题一遍又一遍地弹出:您是否应该将Web APis与客户端库一起 ...
Kafka消费组(consumer group)(转）
转载自:http://www.cnblogs.com/huxi2b/p/6223228.html 一直以来都想写一点关于kafka consumer的东西,特别是关于新版consumer的中文资料很少 ...
Kafka消费组(consumer group)
原文出处:https://www.cnblogs.com/huxi2b/p/6223228.html 转载请注明出处. ---------------------------------------- ...
Java代码测试大端小端
Java果真强大,神马功能的函数都能找到,测试大端小端只需一句话: if(ByteOrder.nativeOrder()==ByteOrder.BIG_ENDIAN)System.out.printl ...
java代码走读,Kafka代码走读-LogManager
https://github.com/haogrgr/haogrgr-test/blob/master/logs/kafka_source.txt 源码阅读(0.8.2.2): (一)概览 1.调用k ...
kafka消费端慢慢延迟（网络带宽不足）
2020-09-29 问题描述:线上业务出现推送延迟,启动测试工具订阅topic,能看到数据正常时间能对上(数据写进去了),用kafka自带的也能对上, 通过分析后发现工具记录的日志在9点41分启动- ...
JAVA代码无效化快捷键_Eclipse常用快捷键及快捷键功能失效时的解决方案
当Eclipse快捷键功能,或者使用菜单时都无效.例如:搜索引用(快捷键Ctrl+Shift+G)无效: 解决方案:转换WorkSpac‍e,或者删除WorkSpace目录下的‍.metadata文件 ...
命令模式 Java代码演示【使用命令模式实现播放器功能】
文章目录模式介绍模式的结构与实现命令模式扩展(实现播放器功能) 模式介绍命令(Command)模式的定义如下:将一个请求封装为一个对象,使发出请求的责任和执行请求的责任分割开.这样俩者之前通过 ...

java代码实现kafka消费端consumer的from-beginning功能

java代码实现kafka消费端consumer的from-beginning功能相关推荐

最新文章

热门文章