Kafka：每个 poll() 调用是否一次只使用一个主题？答案

【问题标题】：Kafka: Have each poll() call only consume from one topic at a time?Kafka：每个 poll() 调用是否一次只使用一个主题？
【发布时间】：2019-04-22 17:08:35
【问题描述】：

我有一个从多个 kafka 主题消费的 Kafka 消费者。我希望能够通过每 100 条消息的 1 次 i/o 调用将批量写入我的目的地，但是为了进行批处理，所有消息都需要来自同一个主题。

如果我有多个主题（比如说 5 个），并且当发生 consumer.poll 或 consumer.consume 时，例如每次轮询我收到 100 条消息，有没有办法确保这些都来自同一个主题，以便可以将这些消息批量写入同一目的地？这样下一个 .poll 调用会获得下一个主题吗？

【问题讨论】：

你想通过这个达到什么目的？
试图变得懒惰，而不必为每个主题生成消费者（我们有数百个主题，但其中一些主题几天没有收到消息，而另一些则每秒收到几个），但在同时具有高性能，因为消息目标数据库对基于主题的批量写入响应更好

标签： apache-kafka

【解决方案1】：

不可能按主题进行轮询 - 您订阅了主题列表，每个主题可能有多个分区。给定的轮询获取ConsumerRecords 对象，该对象是ConsumerRecord 的容器。 ConsumerRecord 代表一个 KV 对，它属于您订阅的主题之一的分区之一。

Kafka 尝试将TopicPartition 分配给基于分配者形成单个组的消费者。如果您只有一个消费者，它将要求所有主题的所有分区。那么没有什么可以阻止您在应用程序代码中进行分组

例如

private void consume() {
    List<String> topics = List.of("topic1", "topic2", "topic3", "topic4", "topic5");
    kafkaConsumer.subscribe(topics);

    while (true) {
        ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(1000);

        topics.forEach(s -> {
            List<ConsumerRecord<String, String>> recordsPerTopicPartition = new ArrayList<>();
            consumerRecords.records(s).forEach(recordsPerTopicPartition::add);
            doWhatever(recordsPerTopicPartition);
        });
    }
}

private void doWhatever(List<ConsumerRecord<String, String>> consumerRecords) {
    //process
}

【讨论】：

【解决方案2】：

另一种处理主题的方法如下：每个ProducerRecord 都有一个方法topic()，它返回该记录的主题名称。然后，您可以按主题分组，并使用该主题的主题对和记录集合做任何您想做的事情。

但如果您想独立处理主题，我强烈建议为每个单独的主题使用单独的 KafkaConsumer。

【讨论】：

【解决方案3】：

在每个主题轮询消息主题时订阅多个主题的一种解决方案是使用暂停/恢复方法。

这是一个例子：

        List<String> subscription = List.of("topic-a", "topic-b");
        consumer.subscribe(suubscription);

        final Map<String, List<TopicPartition>> partitionsPerTopic = 
                consumer.assignment()
                .stream()
                .collect(Collectors.groupingBy(TopicPartition::topic, Collectors.toList()));

        int next = 0;
        consumer.pause(consumer.assignment());
        // Starting consumption
        while (!closed.get()) {
            // Resuming consumption for next topic
            final String topic = subscription.get(next);
            consumer.resume(partitionsPerTopic.get(topic));

            consumer.poll(Duration.ofMillis(500)).forEach( records -> {
            ...
            });   

            // Pausing consumption for current topic
            consumer.pause(partitionsPerTopic.get(topic));
            next = (next + 1) % subscription.size();
       }

但是，此解决方案可能效率不高，因为消费者仍然可以为以前的可提取分区缓冲消息。这些消息将在第二次迭代时被耗尽并再次获取。

换句话说，该解决方案可以增加消费者和代理之间的网络往返。

【讨论】：