KafkaMessageListenerContainer 与 ConcurrentMessageListenerContainer

Question

如果我使用 listener.concurrency 属性，这会使用 KafkaMessageListenerContainer 还是 ConcurrentMessageListenerContainer？请注意，我没有明确定义任何 bean，我将所需的 bean 创建留给 spring boot.

我想知道我的理解是否正确。我希望并行处理来自具有多个分区的主题的消息，比如 3。如果我在 application.yml 文件中如下配置它就可以了。我尝试了以下配置，看到创建了 3 个消费者。

spring:
   kafka:
     consumer:
        group-id: group-id
        key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
        value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
     producer:
        key-serializer: org.apache.kafka.common.serialization.StringSerializer
        value-serializer: org.apache.kafka.common.serialization.StringSerializer
     listener:
        concurrency: 3

所以我的问题是，我是否必须创建一个 bean 来配置 ConcurrentMessageListenerContainer，或者上面的侦听器配置在看到 [= 时会在内部使用 ConcurrentMessageListenerContainer 而不是 KafkaMessageListenerContainer 15=]?

Answer 1

spring.kafka.listener.concurrency= # Number of threads to run in the listener containers.

Concurrency 是 ConcurrentMessageListenerContainer. 的属性如果我们在 application.yml 中定义这个属性意味着我们要使用 ConcurrentMessageListenerContainer.

在Kafka中，一次只允许一个消费者从一个分区读取数据。使用 KafkaMessageListenerContainer 将仅提供单线程消息侦听器消费者。 KafkaMessageListenerContainer

使用ConcurrentMessageListenerContainer将提供在application.yml属性中定义的KafkaMessageListenerContainer的数量。

我们可以使用以下方法获得相同的行为 Listener using Java code.

@Configuration
@EnableKafka
public class KafkaListenerConfiguration {

    @Autowired
    private Environment environment;

    @Bean
    public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.setBatchListener(true);
        return factory;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, environment.getProperty("spring.kafka.bootstrap-servers"));
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "1111");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        return props;
    }
}

我建议您阅读下面的内容 link 以获得相同的实际示例。

https://howtoprogram.xyz/2016/09/25/spring-kafka-multi-threaded-message-consumption/

KafkaMessageListenerContainer 与 ConcurrentMessageListenerContainer

KafkaMessageListenerContainer vs ConcurrentMessageListenerContainer

java

messaging

spring

apache-kafka