如何在特定偏移量到特定偏移量中使用来自 kafka 主题的数据?

问题描述

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

I need to consume specific offset to specific end offset!! consumer.seek() reads the data from specific offset but I need retrieve the data fromoffset to tooffset !! Any help will be appreciate , thanks in advance.

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

推荐答案

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

To read messages from a start offset to an end offset, you first need to use seek() to move the consumer at the desired starting location and then poll() until you hit the desired end offset.

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

For example, to consume from offset 100 to 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;
            }
        }
    }
}

这篇关于如何在特定偏移量到特定偏移量中使用来自 kafka 主题的数据?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！