如果我使用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
?
最佳答案
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/
关于java - KafkaMessageListenerContainer 与 ConcurrentMessageListenerContainer,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/58284866/