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在它看到时使用listener.concurrency=3？

Answer 1

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

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

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

使用将提供在属性中定义ConcurrentMessageListenerContainer的数量。KafkaMessageListenerContainerConcurrencyapplication.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-conspiration/