java - KafkaMessageListenerContainer和ConcurrentMessageListenerContainer

如果我使用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而不是listener.concurrency=3？

最佳答案

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

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



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

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

我们可以使用以下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;
    }
}

我建议您通过下面的链接获得一个实际的例子。

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

关于java - KafkaMessageListenerContainer和ConcurrentMessageListenerContainer，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/58284866/