如何在Spring Boot中配置Kafka消费者并发及自定义属性

嘿，针对你用Spring Boot 1.5.10 + Spring Kafka 1.1.0 + kafka-clients 0.10.0.0-SASL开发消费者时遇到的自定义配置需求，我有几个不用重新定义工厂类的简洁方案，正好适配你的版本：

一、自定义原生Kafka消费者属性（如分区拉取大小、分配策略） #

Spring Boot 1.5.x的自动配置已经支持通过application.properties直接配置大部分原生ConsumerConfig属性，对于那些没有直接暴露的属性，你可以用spring.kafka.consumer.properties.*的前缀来配置，完全不用手动创建ConsumerFactory或者ContainerFactory。

举个例子，配置你提到的两个属性：

# 自定义分区拉取的最大字节数（对应ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG）
spring.kafka.consumer.properties.max.partition.fetch.bytes=10485760

# 自定义分区分配策略（对应ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG）
# 多个策略用逗号分隔，这里用kafka-clients 0.10.x自带的RoundRobinAssignor为例
spring.kafka.consumer.properties.partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor

如果是你自己实现的自定义分配策略，直接写全类名即可。

二、配置消费者并发数 #

Spring Boot 1.5.x没有直接在application.properties里提供全局并发数的配置项，但你可以通过两种简洁方式设置：

1. 全局统一配置 #

不用重新定义整个ConcurrentKafkaListenerContainerFactory，只需要创建一个KafkaListenerContainerFactoryCustomizer的Bean，用来修改默认工厂的并发参数：

import org.springframework.boot.autoconfigure.kafka.KafkaListenerContainerFactoryCustomizer;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.stereotype.Component;

@Component
public class GlobalKafkaContainerCustomizer implements KafkaListenerContainerFactoryCustomizer<ConcurrentKafkaListenerContainerFactory<?, ?>> {

    @Override
    public void customize(ConcurrentKafkaListenerContainerFactory<?, ?> factory) {
        // 设置全局的消费者并发数，比如3个线程
        factory.setConcurrency(3);
    }
}

2. 单个Listener单独配置 #

如果某个消费者需要不同的并发数，直接在@KafkaListener注解里指定concurrency属性即可，这个配置的优先级会高于全局设置：

@KafkaListener(topics = "your-topic", concurrency = "2")
public void consumeMessage(String message) {
    // 消费逻辑
}

补充说明 #

这些方案完全适配你的版本：

• spring.kafka.consumer.properties.*前缀的配置方式在Spring Boot 1.5.x的自动配置中已经支持
• KafkaListenerContainerFactoryCustomizer是Spring Boot 1.5.4及以上版本提供的扩展点，你的1.5.10版本完全兼容

如果有些属性已经在spring.kafka.consumer下有直接的配置项（比如bootstrap-servers、group-id），优先使用直接配置项，没有的再用properties.*的方式覆盖原生属性。

内容的提问来源于stack exchange，提问作者user3842182