我如何读取每个分区的 kafka 偏移量答案

【问题标题】：How i can read kafka offset per partition我如何读取每个分区的 kafka 偏移量
【发布时间】：2020-06-07 10:32:40
【问题描述】：

我不能 for-loop 读取每个分区的 kafka，我不知道我的代码有什么问题，它没有显示我打印的值

示例：我想第一次读取所有偏移分区 0，第二次读取所有偏移分区 1。

（我第一次在stackoverflow上发帖。抱歉交流，希望你能理解我。）

    Consumer<String, String> consumer = new KafkaConsumer<>(props);
    consumer.subscribe(topicNames);
    List<KafkaTopicDataResponse> results = new ArrayList<>();


    try {
        Set<TopicPartition> assignments = consumer.assignment();
        Object[] assignArray = assignments.toArray();
        for (Object topicPartition : assignArray){
            boolean flag = true;
            int receiveRow = 0;
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                if (flag) {
                    consumer.seek((TopicPartition) topicPartition,0);
                    flag = false;
                }
                for (ConsumerRecord<String, String> record : records) {
                    receiveRow++;
                    System.out.printf("offset = %d, partition = %d, key = %s, value = %s%n", record.offset(), record.partition(), record.key(), record.value());
                    logger.info("count : "+ receiveRow);

                }
                if (receiveRow >= limitRow){
                    break;
                }

            }
        }
    } catch(Exception e) {
        logger.error("Exception occured while consuing messages",e);
    }finally {
        consumer.close();
    }

【问题讨论】：

当您执行consumer.poll() 时，您会从分配给该消费者的所有分区中获取消息，而不仅仅是单个分区。
是否有单独读取每个分区的特定用例？
@JavaTechnical 是的，我想读取每个分区

标签： java spring-boot apache-kafka

【解决方案1】：

我想第一次读取所有偏移分区 0，第二次读取所有偏移分区 1。

您说，您想读取分区 0 的所有偏移量，然后读取分区 1。当数据流动即新数据不断到达时，您怎么说 all？

所以当你说 all 时，它应该是 在给定时刻的所有数据。为此，您需要一个一个地分配分区，获取该瞬间的结束偏移量并读取它们直到该结束偏移量。

consumer.assign(Collections.singletonList(new TopicPartition("topic", 0)));

然后你需要获取结束偏移量，因为如果数据不断进入这个分区，这个分区永远不会完成。

TopicPartition tp = new TopicPartition("topic", 0);
long endOffset = consumer.endOffsets(Collections.singletonList(tp)).get(tp);

现在，您必须调用poll() 并检查最后一条记录是否为>= endOffset-1

flag = true;
while(flag) {
       ConsumerRecords records = consumer.poll(Duration.ofSeconds(10));
       for(ConsumerRecord record: records) {
          // process them
          if(record.offset() >= (endOffset-1)) {
             flag = false;
             break;   
          }
       }
}

对其他分区也重复相同的步骤。

【讨论】：