java - 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，或者上面的监听器配置在看到 listener.concurrency=3 时会在内部使用 ConcurrentMessageListenerContainer 而不是 KafkaMessageListenerContainer？

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中Concurrency属性中定义的KafkaMessageListenerContainer数量。

我们可以使用以下使用 Java 代码的 Listener 获得相同的行为。

@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/