我如何在Kafka中使用多个消费者？

Question

我是一个学习Kafka的新学生，我遇到了一些关于理解多个消费者的基本问题，到目前为止，文章、文档等都没有太大的帮助。

我尝试做的一件事是编写我自己的高级Kafka生产者和消费者，并同时运行他们，发布100个简单的消息到一个主题，并让我的消费者检索他们。我成功地做到了这一点，但是当我试图引入第二个消费者来消费刚刚发布消息的同一主题时，它没有收到任何消息。

我的理解是，对于每个主题，您可以有来自不同消费者组的消费者，这些消费者组中的每一个都将获得针对某个主题生成的消息的完整副本。这是正确的吗？如果不是，我设置多个消费者的正确方法是什么？这是我到目前为止写的消费者类：

public class AlternateConsumer extends Thread {
    private final KafkaConsumer<Integer, String> consumer;
    private final String topic;
    private final Boolean isAsync = false;

    public AlternateConsumer(String topic, String consumerGroup) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "localhost:9092");
        properties.put("group.id", consumerGroup);
        properties.put("partition.assignment.strategy", "roundrobin");
        properties.put("enable.auto.commit", "true");
        properties.put("auto.commit.interval.ms", "1000");
        properties.put("session.timeout.ms", "30000");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new KafkaConsumer<Integer, String>(properties);
        consumer.subscribe(topic);
        this.topic = topic;
    }


    public void run() {
        while (true) {
            ConsumerRecords<Integer, String> records = consumer.poll(0);
            for (ConsumerRecord<Integer, String> record : records) {
                System.out.println("We received message: " + record.value() + " from topic: " + record.topic());
            }
        }

    }
}

此外，我注意到，最初我是为了一个只有一个分区的主题'test'来测试上述消耗。当我将另一个消费者添加到一个现有的消费者组，比如“测试组”中时，这触发了Kafka再平衡，它使我的消费延迟大大减慢，以秒为单位。我认为这是一个重新平衡的问题，因为我只有一个分区，但当我创建了一个新主题'MultiplePartitions'（比如6个分区）时，类似的问题出现了，在相同的使用者组中添加更多的使用者会导致延迟问题。我环顾四周，人们告诉我应该使用多线程的消费者--有没有人能解释一下呢？

Answer 1

我认为您的问题出在Auto.offset.reset属性上。当一个新的使用者从一个分区读取，并且没有先前提交的偏移量时，使用auto.offset.reset属性来决定起始偏移量应该是多少。如果您将其设置为“最大”（默认值），您将从最近的（最后的）消息开始读取。如果您将其设置为“最小”，您将得到第一个可用的消息。

因此添加：

properties.put("auto.offset.reset", "smallest");

然后再试一次。

“最小”和“最大”在很久以前就被弃用了。你现在应该用“最早”或“最晚”。有什么问题，查一下文件