Kafka消费者/分区与线程/分区关系

Question

我是 Apache Kafka 的新手，我正在尝试了解以下之间的区别：

1. 创建属于同一组 ID 的两个消费者，它们从同一主题的两个分区消费。
2. 使用两个线程创建一个消费者，这些线程从同一主题的两个分区消费。

在第一种方法中，我真正理解的是每个消费者只会消费“相关”分区的消息，因为这两个消费者属于同一个组。

如下图示例所示，consumer1 将消耗 AAAA 和 BBBB，consumer2 将消耗 CCCC 和 DDDD。

在第二种方法中，正如我在文档中看到的那样：

多线程的消费者：如果处理一条记录需要一段时间，单个消费者可以运行多个线程来处理记录，但很难管理每个线程/任务的偏移量。如果一个消费者运行多个线程，那么同一分区上的两条消息可能由两个不同的线程处理，这使得在没有复杂的线程协调的情况下很难保证记录传递顺序。如果处理单个任务需要很长时间，此设置可能是合适的，但请尽量避免。

所以在下面的例子中，可能会发生一些不同的情况：

• Thread1 消耗 AAAA 和 CCCC / Thread2 消耗 BBBB 和 DDDD
• Thread1 消耗 AAAA 和 BBBB / Thread2 消耗 CCCC 和 DDDD
• Thread1 消耗 AAAA 和 DDDD / Thread2 消耗 CCCC 和 BBBB
• Thread1 消耗 CCCC 和 DDDD / Thread2 消耗 AAAA 和 BBBB
• Thread1 消耗 BBBB 和 DDDD / Thread2 消耗 AAAA 和 CCCC
• Thread1 消耗 BBBB 和 CCCC / Thread2 消耗 AAAA 和 DDDD

这两种方法之间的区别看起来微不足道。而在第一种方法中，每个消费者只按顺序消费分配给它的分区，在第二种方法中，每个线程都可以从两个分区消费，但可能会发生 Thread1 消费 AAAA 和 Thread2 BBBB 并且 BBBB 的进程在进程 AAAA 之前完成结束了。

我用 springboot 开发了一个 Kafka 消费者。我的 kafka 主题有 5 个分区。每个分区都有需要按顺序使用的信息。我决定只创建一个具有 5 个线程的消费者，而不是创建 5 个消费者。从理论上讲，这并没有什么意义，因为我需要按顺序使用每个分区的信息，而创建线程会破坏该逻辑。​​

在 Kafka 侦听器中，我定义了 RECEIVED_PARTITION_ID 以打印从哪个分区使用数据。我已将 log4j 配置为打印运行 KafkaListener 的线程，例如：[org.springframework.kafka.KafkaListenerEndpointContainer#0-thread-C-1]

KafkaConsumer.java的代码：

@KafkaListener(topics = "${kafka.topic}", groupId = "${kafka.groupid}", containerFactory = "kafkaListenerContainerFactory")    
public void consumeJson(String mensaje,
            @Header(KafkaHeaders.RECEIVED_PARTITION_ID) String partitionID) throws InterruptedException {
        logger.info("-> Consumed JSon message");
        logger.info("-> Read from partition:" + partitionID);
        logger.info("-> Consumed JSON Message: [" + mensaje.toString()+"]");
}

KafkaConfiguration.java的代码：

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    Map<String, Object> config = new HashMap<>();

    config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    config.put(ConsumerConfig.GROUP_ID_CONFIG, groupid);
    config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPoll);
    // Commit manual (para poner AckMode.BATCH), por configuración
    config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, autoCommit);

    return new DefaultKafkaConsumerFactory<>(config);
}

@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String,String>> kafkaListenerContainerFactory(){
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<String, String>();
    factory.setConsumerFactory(consumerFactory());
    factory.setConcurrency(5);
    // Commit the offset when the listener returns after processing the record. 
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.BATCH);
    return factory;
}

当我运行我的 springboot 应用程序并看到日志时，看起来一个线程/分区关系已经完成，就像创建了一个消费者/分区一样。每个线程只消耗一个分区的数据！线程 0 -> 分区 0，线程 1 -> 分区 1，线程 2 -> 分区 2，线程 3 -> 分区 3，线程 4 -> 分区 4。我检查了很多 Log 行，它总是匹配的。

日志的一小部分：

[org.springframework.kafka.KafkaListenerEndpointContainer#0-3-C-1][9980][e.e.s.l.KafkaConsumer:][][] 消费的 JSON 消息 从 partition3 读取

[org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1][9980][e.e.s.l.KafkaConsumer:][][] 消费的 JSON 消息 从分区 1 读取

[org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1][9980][e.e.s.l.KafkaConsumer:][][] 消费的 JSON 消息 从 partition0 读取

[org.springframework.kafka.KafkaListenerEndpointContainer#0-4-C-1][9980][e.e.s.l.KafkaConsumer:][][] 消费的 JSON 消息 从 partition4 读取

[org.springframework.kafka.KafkaListenerEndpointContainer#0-2-C-1][9980][e.e.s.l.KafkaConsumer:][][] 消费 JSON 消息 从 partition2 读取

[org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1][9980][e.e.s.l.KafkaConsumer:][][] 消费的 JSON 消息 从分区 1 读取

这对我来说没有意义。正如文档所说：

多线程的消费者：如果处理一条记录需要一段时间，单个消费者可以运行多个线程来处理记录，但很难管理每个线程/任务的偏移量。如果一个消费者运行多个线程，那么同一分区上的两条消息可能由两个不同的线程处理，这使得在没有复杂的线程协调的情况下很难保证记录传递顺序。如果处理单个任务需要很长时间，此设置可能是合适的，但请尽量避免。

谁能解释一下发生了什么？ 同一个消费者组5个消费者/5个分区关系和1个消费者/5个线程关系有什么区别？

现在看起来是一样的。

谢谢！

Answer 1

你不应该执行异步处理；使用 5 个消费者，每个消费者获得一个分区，是保证有序处理的正确方法和唯一方法。

使用来自单个消费者的多个线程会导致偏移管理问题。

将并发设置为 5 会创建 5 个消费者。

这对我来说没有意义。正如文档所说：

哪个“文档”？

最后一句是对的：

如果处理单个任务需要很长时间，此设置可能是合适的，但请尽量避免。

重要的词是might；但我会说永远不会。