为多个主题创建一个Kafka消费者

Question

我想为几个主题创建一个kafka消费者。consumer的方法构造函数允许我在订阅中传输主题列表的参数，如下所示：

private Consumer createConsumer() {
    Properties props = getConsumerProps();
    Consumer<String, byte[]> consumer = new KafkaConsumer<>(props);
    ArrayList<String> topicMISL = new ArrayList<>();
    for (String s:Connect2Redshift.kafkaTopics) {
        topicMISL.add(systemID + "." + s);
    }
    consumer.subscribe(topicMISL);
    return consumer;
}


private boolean consumeMessages( Duration duration, Consumer<String, byte[]> consumer) {
        try {  Long start = System.currentTimeMillis();
            ConsumerRecords<String, byte[]> consumerRecords = consumer.poll(duration);
            }
   }

之后，我想轮询记录从Kafka流每3秒并处理它们，但我想知道什么是这个消费者-如何将不同主题的记录轮询-首先一个主题，然后另一个，或并行。会不会一个消息量大的主题会一直处理，另一个消息量小的主题会等待？

Answer 1

ConsumerRecords<String, String> records = consumer.poll(long value);
    for (TopicPartition partition : records.partitions()) {
        List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
        for (ConsumerRecord<String, String> record : partitionRecords) {
            
        }
        
    }

还需要通过查找offset和使用consumer进行commit for offset。委员会同步

Answer 2

一般来说，这取决于您的主题设置。kafka通过使用每个主题的多个分区来缩放。

• 如果一个主题有3个分区，Kafka可以并行读取
• 对于多个主题也是如此，阅读可以并行进行

如果您有一个分区接收的消息比其他分区多得多，那么您可能会遇到这个特定分区的使用者延迟的情况。调整批量大小和消费者设置可能会对他们有所帮助，同时也会压缩消息。理想情况下，确保均匀分布负载可以避免这种情况。

看看这篇博客文章，它让我对内在有了很好的理解：https://www.confluent.io/blog/configure-kafka-to-minimize-latency/