Kafka Consumer poll() 方法被阻塞答案

【问题标题】：Kafka Consumer's poll() method gets blockedKafka Consumer poll() 方法被阻塞
【发布时间】：2016-10-07 12:41:57
【问题描述】：

我是 Kafka 0.9 的新手，在测试一些功能时，我发现 Java 实现的 Consumer (KafkaConsumer) 中出现了一个奇怪的行为。

Kafka 代理位于 Ambari 外部机器中。

即使您可以实现生产者并开始向外部代理发送消息，我也不知道为什么当消费者尝试读取事件（轮询）时，它会卡住。

我知道生产者运行良好，因为我确实可以通过控制台消费者（在 ambari 本地工作）来消费消息。但是当我执行 Java Consumer 时，什么都没有发生，只是卡住了。调试代码我可以看到它在poll() 行被阻止：

    ConsumerRecords<String, String> records = consumer.poll(100);

顺便说一句，超时没有任何作用。无论您输入 0、100 还是 1000 毫秒，消费者都会在这一行中被阻塞，不会超时也不会抛出异常。

我尝试了各种替代属性，例如 advertised.host.name、advertised.listener 等等，但运气为零。 p>

任何帮助将不胜感激。提前致谢！

【问题讨论】：

您能否以不同的方式使用这些消息，例如使用kafka-console-consumer.sh ？
是的，我是。在托管 ambari 的机器上，我可以通过控制台消费者使用消息
那么在您运行消费者的机器上呢？您在那里尝试过控制台消费者吗？
我没有，因为我没有安装kafka，也没有安装zookeeper。
你不需要在那里安装 Zookeeper，只需将 Kafka 二进制文件解压到某个地方。如果您想排除网络连接问题（防火墙等），那么您几乎必须这样做。否则你无法解决这类问题。您的问题可能很简单，因为您的消费者由于防火墙问题而无法连接到您的 Zookeeper 实例。

标签： apache-kafka polling kafka-consumer-api kafka-producer-api

【解决方案1】：

原因可能是运行消费者代码的机器无法连接到 Zookeeper。尝试在安装了 Kafka 的机器上运行相同的消费者代码（我试过这个并为我工作）。我还通过在server.properties 文件中提及以下属性解决了这个问题：

advertised.host.name="ip address which you want to expose"

// In my case, it is the public IP of the EC2 machine, I have kafka and zookeeper installed on the same EC2 machine.

advertised.port=9092

关于声明：

ConsumerRecords<String, String> records = consumer.poll(100);

上面的陈述并不意味着消费者会在 100 毫秒后超时；相反，这是轮询期。它在 100 毫秒内捕获的任何数据都会被读入记录集合。

【讨论】：

> 它在 100 毫秒内捕获的任何数据都会被读入记录集合。这不一定是正确的。当没有可立即使用的记录时，超时是轮询时间的上限。来自 javadoc (KafkaConsumer.poll(Duration timeout))：``` 如果有可用记录，此方法会立即返回。否则，它将等待通过的超时。如果超时，将返回一个空记录集。请注意，此方法可能会在超时后阻塞以执行自定义 {@link ConsumerRebalanceListener} 回调。 ```
就我而言，即使我使用的是 (KafkaConsumer.poll(Duration timeout));它有时会陷入无限循环并且不会出来。不知道如何使它更加健壮和稳定。有什么想法吗？

【解决方案2】：

在我的情况下，poll() 方法最终陷入无限循环 ensureCoordinatorReady()，Coordinator 一词提到我协调器在另一台主机上运行。（出于测试目的，我只将一个代理主机添加到我的 /etc /hosts 虽然总共有三个经纪人）。所以消费者正确地得到消费者协调员。

所以解决方案出来了：在 /etc/hosts 文件中配置正确运行 kafka 代理的主机

【讨论】：