如何使用spring boot设置kafka使用者并发性

Question

我正在编写一个基于Java的Kafka消费者应用程序。我正在为我的应用程序使用kafka-clients、Spring Kafka和Spring boot。虽然Spring boot让我可以轻松地编写Kafka消费者（无需真正编写ConcurrentKafkaListenerContainerFactory、ConsumerFactory等），但我希望能够为这些消费者定义/定制一些属性。然而，我无法找到一个简单的方法来做它使用Spring Boot。我有兴趣设置的一些属性是-

consumerconfig.max_partition_fetch_bytes_configconsumerconfig.partition_assignment_strategy_config

我在这里查看了Spring Boot的预定义属性。

版本-

• Kafka-客户机-0.10.0.0-sasl
• Spring-Kafka-1.1.0.发布
• Spring Boot-1.5.10.发布

Answer 1

在您引用的URL处，向下滚动到

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

Spring-Kafka-1.1.0.发行版

我建议至少升级到1.3.5；它有一个简单得多的线程模型，这要归功于KIP-62。

spring.kafka.consumer.properties.heartbeat.interval.ms

@Bean
public ConsumerFactory<?, ?> kafkaConsumerFactory(KafkaProperties properties) {
    Map<String, Object> consumerProps = properties.buildConsumerProperties();
    consumerProps.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 5_000);
    return new DefaultKafkaConsumerFactory<Object, Object>(consumerProps);
}