5

我使用的是 kafka-0.8.1.1 版本，在这个版本中，自动重新连接不能通过 java 代码工作。我有属性文件

props.put("zookeeper.connect", "zoo1:2181,zoo2:2181");
props.put("group.id", KAFKA_GROUPID);
props.put("zookeeper.session.timeout.ms", "10000");
props.put("zookeeper.sync.time.ms", "200");
props.put("auto.commit.interval.ms", "1000");

当我启动消费者线程时，它已连接到 kafka 服务器，在运行服务器的线程之间失去与 kafka 服务器的连接 2 小时。如果它重新连接 kafka 消费者没有收到任何消息，并且它也不会在流动。

我们如何检测它丢失连接并自动重新连接消费者线程。

提前致谢

4

0 回答 0