我有一個(gè)具有多個(gè)分區(qū)的 Kafka 主題，我想知道 Java 中是否有一種方法可以獲取該主題的最后一條消息。我不關(guān)心分區(qū)，我只想獲取最新消息。我已經(jīng)嘗試過@KafkaListener，但它僅在主題更新時(shí)才獲取消息。如果應(yīng)用程序打開后沒有發(fā)布任何內(nèi)容，則不會(huì)返回任何內(nèi)容。也許傾聽者根本就不是解決問題的正確方法？
2 回答
牛魔王的故事
TA貢獻(xiàn)1830條經(jīng)驗(yàn) 獲得超3個(gè)贊
以下片段對(duì)我有用。你可以試試這個(gè)。評(píng)論里有解釋。
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貢獻(xiàn)2011條經(jīng)驗(yàn) 獲得超2個(gè)贊
您必須使用每個(gè)分區(qū)中的最新消息，然后在客戶端進(jìn)行比較（使用消息上的時(shí)間戳，如果包含它）。原因是 Kafka 不保證分區(qū)間的順序。在分區(qū)內(nèi)，您可以確定偏移量最大的消息是最新推送到該分區(qū)的消息。
添加回答
舉報(bào)
0/150
提交
取消