Kafka High Level Consumer 关闭线程答案

【问题标题】：Kafka High Level Consumer shuts down threadKafka High Level Consumer 关闭线程
【发布时间】：2014-04-12 18:21:24
【问题描述】：

我使用的是高级消费者，如：https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example

我注意到我的消费者不会永远运行并会在一段时间后结束。在 zookeeper 端，我看到以下内容：

INFO sessionid 的已处理会话终止：0x144a4854325004d (org.apache.zookeeper.server.PrepRequestProcessor)
信息已关闭客户端 /127.0.0.1:59899 的套接字连接 sessionid 0x144a4854325004d (org.apache.zookeeper.server.NIOServerCnxn)

我正在使用默认配置。如何让我的消费者永远倾听？

【问题讨论】：

您是否尝试更改zookeeper.session.timeout.ms、zookeeper.sync.time.ms 参数以查看是否对此有任何影响.. 并且查看更多堆栈跟踪以及您的消费者代码会很有用正在运行。
你解决了吗..我也面临同样的问题。 :(
这里没有足够的信息来弄清楚发生了什么。 Zookeeper 超时可能是个问题。也许你陷入了重新平衡循环？

标签： apache-zookeeper apache-kafka

【解决方案1】：

您可能想尝试运行 ConsoleConsumer。

【讨论】：

控制台消费者运行良好。我想对消息做一些进一步的处理——所以必须通过代码来做。有什么想法吗？
你是处理生产者端还是消费者端？可以与控制台消费者一起使用。
消费者方面。我认为使用java代码会更容易。 wiki 说“这里有趣的部分是 while (it.hasNext()) 部分。基本上这段代码从 Kafka 读取，直到你停止它。”我不确定这是否准确。
您可以重写 ConsoleConsumer 以插入您想要更改消息的任何操作。另一种可能性是消息更改本身是您断开连接的原因 - 可能值得检查。
@La-comadreja 对于消费者来说，没有所谓的“所有消息”。它是一个流，因此消费者无法先验地知道将要发送多少条消息。我使用了相同的代码，它在hasNext() 调用中阻塞，直到收到新消息。 @user3399291，请发布更多日志消息和周围的代码。可用的信息越多，就越容易发现问题。