Kafka Consumer 需要较长的轮询持续时间答案

【问题标题】：Kafka Consumer needs a long poll durationKafka Consumer 需要较长的轮询持续时间
【发布时间】：2020-06-27 05:21:23
【问题描述】：

使用具有以下配置的 Kafka/Java：

props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers);
props.put(ConsumerConfig.GROUP_ID_CONFIG, this.groupId);
props.put(ConsumerConfig.CLIENT_ID_CONFIG, UUID.randomUUID().toString());
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, this.maxPollRecords);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializerClass.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializerClass.getName());
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_COMMITTED.toString().toLowerCase(Locale.ROOT));

我有一个简单的轮询循环，例如：

consumer.poll(Duration.ofMillis(200));

我注意到一些奇怪的行为。持续时间为 0，它不返回任何结果。在本地，持续时间为 200 毫秒，我得到了一些结果，但在另一个生产环境中它永远不会返回结果，它至少需要 1 秒。

在我的理解中，poll 方法会等到至少找到一个结果。持续时间为0，至少应该返回已经到达的结果，而不应该总是不返回任何结果。

解释是什么？

【问题讨论】：

标签： java apache-kafka kafka-consumer-api

【解决方案1】：

根据文档：

public ConsumerRecords<K,V> poll(long timeout)
timeout - The time, in milliseconds, spent waiting in poll if data is not available in the buffer. If 0, returns immediately with any records that are available currently in the buffer, else returns empty. Must not be negative.

因此，基本上作为轮询请求阻塞正在调用它的线程，轮询持续时间是它可以阻塞线程的最长时间。因此，如果超时时间为零或小于发出请求并获得消费者响应所需的时间，则不会返回任何记录。

仅供参考，如果我们将超时设置为高并将消费者的 max.poll.records 属性设置为我们想要的值，假设 max.poll.records : "10" ，那么轮询将在获取 10 条记录后自行结束（即使超时很大）。所以理想情况下需要知道网络延迟，否则我上面提到的技巧可以正常工作。

【讨论】：

【解决方案2】：

主要问题是您可能没有存储poll() 的输出。

ConsumerRecords records = consumer.poll(10000);

您需要将poll 的结果存储在ConsumerRecords 中。您可以找到示例代码-sn-ps here

【讨论】：

这只是一个例子。事实上，我将结果存储在一个变量中以使用它们（并且它可以工作，具有足够长的轮询持续时间）。
如果您的 Kafka 集群已配置大量最小同步副本，并且生产者配置为 ack=all，那么一旦所有最小同步副本都收到消息，消费者就可以使用数据。这可能需要一些时间。
那么请告诉我您的集群是如何设计的。 Kafka 建议在数据中心内设置集群。不建议集群跨越不同的数据中心。你的生产者和 kafka 在同一个数据中心吗？