时间戳

首页 > 解决方案 > 如何将特定偏移量中来自kafka主题的数据消耗到特定偏移量？

java - 如何将特定偏移量中来自kafka主题的数据消耗到特定偏移量？

问题描述

我需要将特定的偏移量消耗到特定的结束偏移量！！consumer.seek() 从特定偏移量读取数据，但我需要从偏移量检索数据到 tooffset ！任何帮助将不胜感激，在此先感谢。

    ConsumerRecords<String, String> records = consumer.poll(100);
    if(flag) {
        consumer.seek(new TopicPartition("topic-1", 0), 90);
        flag = false;
    }

标签： javaapache-kafkakafka-consumer-api

解决方案

要从起始偏移量读取消息到结束偏移量，您首先需要使用seek()将消费者移动到所需的起始位置，然后poll()直到您到达所需的结束偏移量。

例如，从偏移量 100 到 200 消耗：

String topic = "test";
TopicPartition tp = new TopicPartition(topic, 0);

try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(configs)) {
    consumer.subscribe(Arrays.asList(topic), new ConsumerRebalanceListener() {
        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {}

        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
            // Move to the desired start offset 
            consumer.seek(tp, 100L);
        }
    });
    boolean run = true;
    long lastOffset = 200L;
    while (run) {
        ConsumerRecords<String, String> crs = consumer.poll(Duration.ofMillis(100L));
        for (ConsumerRecord<String, String> record : crs) {
            System.out.println(record);
            if (record.offset() == lastOffset) {
                // Reached the end offsey, stop consuming
                run = false;
                break;
            }
        }
    }
}

推荐阅读