12-09
12-09
12-09
12-09
12-09
12-09
12-09
12-09
12-09
12-09
12-09
12-09
ADADADADAD
电脑知识 时间:2024-12-03 14:47:24
作者:文/会员上传
12-09
12-09
12-09
12-09
12-09
12-09
12-09
12-09
12-09
12-09
12-09
12-09
Kafka可以通过设置consumer的offset来读取指定位置的消息。在创建consumer实例时,可以通过指定partition和offset来设置consumer的起始位置。具体步骤如下:创建Kafka consumer
以下为本文的正文内容,内容仅供参考!本站为公益性网站,复制本文以及下载DOC文档全部免费。
Kafka可以通过设置consumer的offset来读取指定位置的消息。在创建consumer实例时,可以通过指定partition和offset来设置consumer的起始位置。具体步骤如下:
auto.offset.reset
属性为none
,禁止consumer自动重置offset。这样可以确保consumer从指定的offset开始读取消息。Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("group.id", "test-group");props.put("enable.auto.commit", "false");props.put("auto.offset.reset", "none");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
assign()
方法将consumer分配到指定的partition,并设置起始offset。TopicPartition partition = new TopicPartition("test-topic", 0);consumer.assign(Collections.singletonList(partition));consumer.seek(partition, 10); // 从offset为10的位置开始读取消息
poll()
方法来获取消息了。ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));for (ConsumerRecord<String, String> record : records) {System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());}
通过以上步骤,就可以在Kafka中读取指定位置的消息。
11-20
11-19
11-20
11-20
11-20
11-19
11-20
11-20
11-19
11-20
11-19
11-19
11-19
11-19
11-19
11-19