Kafka高级消费者

Question

我正在尝试使用高级消费者批量读取Kafka主题中的消息。在这批读取期间，我的线程必须在某个时候停止。

或者，一旦主题中的所有消息都用完了。或获取消息即将被读取时的最大偏移量，并停止直到达到最大偏移量。

我尝试在高级消费者处使用代码，但 KafkaStream 上的迭代器方法似乎是一个阻塞调用，并等待另一条消息传入。

所以3个问题，

>

我怎么知道没有更多消息要从该主题中读取？

如果我对上述问题有答案，我该如何阻止它再听这个话题？

有没有办法找到批量读取开始时的最大偏移量(我认为简单的消费者可以做到这一点)，让高级消费者在那个点停下来？

Answer 1

您可以选择在指定时间内没有新消息到达时，将所有消息视为已读。这可以使用消费者属性consumer.timeout.ms进行设置。在该指定值通过而没有任何新消息到达后，消费者迭代器将抛出超时异常，您可以在消费者中处理该异常并退出。