Kafka - 在 Java 中聆听主题的比 poll() 更好的替代方法是什么？

Question

我正在尝试用 Java 创建一个消费者客户端。我意识到 poll() 函数已贬值。听卡夫卡的话题有哪些选择？

我的代码：

KafkaConsumer< String, UserSegmentPayload > kc = new KafkaConsumer<>(props2);
kc.subscribe(Collections.singletonList(topicName));
while (true) {
    ConsumerRecords<String, UserSegmentPayload> records = kc.poll(100);
    for (ConsumerRecord<String, UserSegmentPayload> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s\n",
        record.offset(), record.key(), record.value());
    }
}

Answer 1

的原因poll()和poll(long)被弃用是它们可能无限期阻塞（即使在第二情况下，当指定了超时）。此行为的根本原因是这些方法中的初始元数据更新可能会永远阻塞（请参阅此处）。相反，你应该使用poll(Duration)的-方法KafkaConsumer。所以，在你的代码，你需要做的是，以取代kc.poll(100)用kc.poll(Duration.ofMillis(100))。