Java代码：使用特定主题的kafka使用者检索最后n条消息

Question

kafka版本：0.9.0.1

如果n = 20，我必须得到一个主题的最后20条消息。

现在，我正在使用

kafkaConsumer.seekToBeginning();

我收到所有消息。我正在写一些逻辑，以获得最后20。

我的主题可能有数十万条记录

public List<JSONObject> consumeMessages(String kafkaTopicName) {
  KafkaConsumer<String, String> kafkaConsumer = null;
  boolean flag = true;
  List<JSONObject> messagesFromKafka = new ArrayList<>();
  int recordCount = 0;
  int i = 0;
  int maxMessagesToReturn = 20;

  Properties props = new Properties();         
  props.put("bootstrap.servers", "localhost:9092");
  props.put("group.id", "project.group.id");
  props.put("max.partition.fetch.bytes", "1048576000");
  props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  kafkaConsumer = new KafkaConsumer<>(props);

  kafkaConsumer.subscribe(Arrays.asList(kafkaTopicName));
  TopicPartition topicPartition = new TopicPartition(kafkaTopicName, 0);
  LOGGER.info("Subscribed to topic " + kafkaConsumer.listTopics());
  while (flag) {
    // will consume all the messages and store in records
    ConsumerRecords<String, String> records = kafkaConsumer.poll(1000);

    kafkaConsumer.seekToBeginning(topicPartition);

    // getting total records count
    recordCount = records.count();
    LOGGER.info("recordCount " + recordCount);
    for (ConsumerRecord<String, String> record : records) {
      if(record.value() != null) {
        if (i >= recordCount - maxMessagesToReturn) {
          // adding last 20 messages to messagesFromKafka
          LOGGER.info("kafkaMessage "+record.value());
          messagesFromKafka.add(new JSONObject(record.value()));
        }
        i++;
      }
    }
    if (recordCount > 0) {
      flag = false;
    }
  }
  kafkaConsumer.close();
  return messagesFromKafka;
}

Answer 1

您可以使用kafkaConsumer.seekToEnd(Collection<TopicPartition> partitions)来寻找给定分区的最后一个偏移量。根据文件：

“寻找每个给定分区的最后一个偏移量。这个函数懒惰地评估，只有在调用poll(Duration)或position(TopicPartition)时才在所有分区中寻找最终偏移量。如果没有提供分区，则寻找当前所有分区的最终偏移量分配的分区。“

然后，您可以使用position(TopicPartition partition)检索特定分区的位置。

然后你可以从它减少20，并使用kafkaConsumer.seek(TopicPartition partition, long offset)来获取最近的20条消息。

只是，

kafkaConsumer.seekToEnd(partitionList);
long endPosition = kafkaConsumer.position(topicPartiton);
long recentMessagesStartPosition = endPosition - maxMessagesToReturn;
kafkaConsumer.seek(topicPartition, recentMessagesStartPosition);

现在，您可以使用poll()检索最近的20条消息

这是一个简单的逻辑，但是如果你有多个分区，你也必须考虑这些情况。我没试过这个，但希望你能得到这个概念。