卡夫卡高级消费者

Question

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

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

我尝试使用high-level-consumer 的代码，但 KafkaStream 上的迭代器方法似乎是一个阻塞调用，并等待其他消息进来。

所以 3 个问题，

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

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

3. 有没有办法在批量读取开始时找到最大偏移量（我认为简单的消费者可以做到这一点）并让高级消费者在该点停止？

Answer 1

您可以选择在指定时间段内没有新消息到达时决定是否已阅读所有消息。这可以使用消费者属性consumer.timeout.ms 进行设置。在此指定值过去后没有任何新消息到达，ConsumerIterator 将抛出超时异常，您可以在消费者中处理并退出。