单个主题的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();
}

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

是不是就这么简单，把factory.setConcurrency(5);加到我的postKafkaListenerContainerFactory()上，比如：

@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

这不是 Apache Kafka 的工作方式。一个想法总是在单个线程中的同一分区中存在进程记录。 factory.setConcurrency(5); 肯定是围绕一个主题中有多少个分区。所以，如果你只有一个，这个属性不会带来任何价值。如果主题中有 10 个分区，那么 Spring Kafka 会产生 5 个线程，每个线程将处理 2 个分区。

我想说这在Reference Manual 中很清楚：

例如，如果提供了 6 个 TopicPartition 并且并发为 3；每个容器将获得 2 个分区。 5个TopicPartition，2个容器分2个分区，第3个分1个。如果并发大于TopicPartition个数，则调低并发，每个容器分1个分区。

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