Kafka使用者/分区与线程/分区关系

Question

我对Apache Kafka是新手，我试图理解以下两个方面的区别：

1. 创建属于同一组id的两个使用者，这些使用者来自同一主题的两个分区。
2. 用两个线程创建一个使用者，这些线程来自同一主题的两个分区。

在第一种方法中，我实际上理解的是，每个使用者将只使用与之“相关”的分区的消息，因为这两个使用者属于同一个组。

因此，在下面的示例中，可能会发生一些不同的情况：

• 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完成之前就完成了。

@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()+"]");
}

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应用程序并看到日志时，它看起来就像是完成了一个线程/分区关系，就像是创建了一个使用者/分区一样。每个线程只消耗来自一个分区的数据！Thread0->Partition0、Thread1->Partition1、Thread2->Partition2、Thread3->Partition3、Thread4->Partition4。我已经检查了日志行的数组，它总是匹配的。

日志的一小部分：

[org.springframework.kafka.kafkalistenerEndpointContainer#0-0-C-1][9980][E.E.S.L.KafkaConsumer:][][]从分区读取的已消费JSon消息0

[org.springframework.kafka.kafkalistenerEndpointContainer#0-4-C-1][9980][E.E.S.L.Kafkaconsumer:][][]从分区读取的已消费的JSon消息4

[org.springframework.kafka.kafkalistenerEndpointContainer#0-2-C-1][9980][E.E.S.L.KafkaConsumer:][][]从分区读取的已消费JSon消息2

[org.springframework.kafka.kafkalistenerEndpointContainer#0-1-C-1][9980][E.E.S.L.KafkaConsumer:][][]从分区读取的已消费JSon消息1

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

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

谢谢！

Answer 1

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

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

将并发设置为5将创建5个使用者。