Spring Kafka单个主题的多个消费者消耗不同的消息

Question

在我的Spring Boot Kafka应用程序中，我有以下使用者配置：

@Bean
public ConsumerFactory<String, Post> postConsumerFactory(KafkaProperties kafkaProperties) {
    return new DefaultKafkaConsumerFactory<>(kafkaProperties.buildConsumerProperties(), new StringDeserializer(), new JsonDeserializer<>(Post.class));
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Post> postKafkaListenerContainerFactory(KafkaProperties kafkaProperties) {

    kafkaProperties.getProperties().put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, kafkaConsumerMaxPollIntervalMs);
    kafkaProperties.getProperties().put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaConsumerMaxPollRecords);

    ConcurrentKafkaListenerContainerFactory<String, Post> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
    factory.setConsumerFactory(postConsumerFactory(kafkaProperties));

    return factory;
}

消费者：

@KafkaListener(topics = "${kafka.topic.post.send}", containerFactory = "postKafkaListenerContainerFactory")
public void sendPost(ConsumerRecord<String, Post> consumerRecord, Acknowledgment ack) {

    // do some logic

    ack.acknowledge();
}

如果我理解正确的话，现在我有一个消费者的实例。我想增加post消费者的数量，假设有5个消费者将消费来自${kafka.topic.post.send}的不同（不同）消息，以加快消息消费。

它是否像添加工厂一样简单。setConcurrency（5） 至我的PostKafkAlisterContainerFactory（），例如：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Post> postKafkaListenerContainerFactory(KafkaProperties kafkaProperties) {

    kafkaProperties.getProperties().put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, kafkaConsumerMaxPollIntervalMs);
    kafkaProperties.getProperties().put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaConsumerMaxPollRecords);

    ConcurrentKafkaListenerContainerFactory<String, Post> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
    factory.setConsumerFactory(postConsumerFactory(kafkaProperties));
    factory.setConcurrency(5);

    return factory;
}

或者我需要做一些额外的工作来实现它吗？

Answer 1

阿帕奇·Kafka不是这样工作的。一个想法是，在单个线程的同一个分区中始终存在进程记录。那家工厂。setConcurrency（5） 肯定是关于一个主题中有多少个分区。所以，如果你只有一个，这个属性不会带来任何价值。如果主题中有10个分区，那么Spring Kafka会生成5个线程，每个线程将处理2个分区。

我想说的是，这在参考手册中非常清楚：

假设提供了6个TopicPartition，并发性为3；每个容器将获得2个分区。对于5个TopicPartition，2个容器将获得2个分区，第三个容器将获得1个分区。如果并发性大于TopicPartitions的数量，那么并发性将被下调，这样每个容器将获得一个分区。

因此，如果您想拥有所描述的这种并发性，您确实需要在主题中创建5个分区。只有在这之后，您才能并行处理同一主题中的记录。