我有一個具有多個分區的 Kafka 主題，我想知道 Java 中是否有一種方法可以獲取該主題的最后一條消息。我不關心分區，我只想獲取最新消息。我已經嘗試過@KafkaListener，但它僅在主題更新時才獲取消息。如果應用程序打開后沒有發布任何內容，則不會返回任何內容。也許傾聽者根本就不是解決問題的正確方法？
2 回答
牛魔王的故事
TA貢獻1830條經驗 獲得超3個贊
以下片段對我有用。你可以試試這個。評論里有解釋。
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
consumer.subscribe(Collections.singletonList(topic));
consumer.poll(Duration.ofSeconds(10));
consumer.assignment().forEach(System.out::println);
AtomicLong maxTimestamp = new AtomicLong();
AtomicReference<ConsumerRecord<String, String>> latestRecord = new AtomicReference<>();
// get the last offsets for each partition
consumer.endOffsets(consumer.assignment()).forEach((topicPartition, offset) -> {
System.out.println("offset: "+offset);
// seek to the last offset of each partition
consumer.seek(topicPartition, (offset==0) ? offset:offset - 1);
// poll to get the last record in each partition
consumer.poll(Duration.ofSeconds(10)).forEach(record -> {
// the latest record in the 'topic' is the one with the highest timestamp
if (record.timestamp() > maxTimestamp.get()) {
maxTimestamp.set(record.timestamp());
latestRecord.set(record);
}
});
});
System.out.println(latestRecord.get());
森林海
TA貢獻2011條經驗 獲得超2個贊
您必須使用每個分區中的最新消息，然后在客戶端進行比較（使用消息上的時間戳，如果包含它）。原因是 Kafka 不保證分區間的順序。在分區內，您可以確定偏移量最大的消息是最新推送到該分區的消息。
添加回答
舉報
0/150
提交
取消