如何在Spring Kafka中为不同主题配置专属消费者属性

不用自定义ContainerFactory的话，你可以结合Spring Boot的application.properties和@KafkaListener的properties属性来实现每个主题的独立消费者配置，具体步骤如下：

1. 在application.properties中按主题分组配置属性 #

首先把每个主题的专属消费者属性用独立前缀区分开，同时可以保留公共属性在spring.kafka.consumer下（所有消费者会继承这些公共配置）：

# 公共消费者配置（所有主题共享）
spring.kafka.consumer.auto-offset-reset=latest

# 主题topic1的专属配置
topic1.consumer.group-id=test-group-1
topic1.consumer.key-deserializer=com.example.MyKeyDeserializer
topic1.consumer.value-deserializer=com.example.MyValueDeserializer
topic1.consumer.spring.deserializer.key.delegate.class=com.example.MyKeyDelegateDeserializer
topic1.consumer.spring.deserializer.value.delegate.class=com.example.MyValueDelegateDeserializer

# 主题topic2的专属配置
topic2.consumer.group-id=test-group-2
topic2.consumer.key-deserializer=com.example.AnotherKeyDeserializer
topic2.consumer.value-deserializer=com.example.AnotherValueDeserializer
topic2.consumer.spring.deserializer.key.delegate.class=com.example.AnotherKeyDelegateDeserializer
topic2.consumer.spring.deserializer.value.delegate.class=com.example.AnotherValueDelegateDeserializer

2. 在@KafkaListener中引用对应主题的配置 #

通过@KafkaListener的properties属性，使用SpEL表达式读取配置文件中对应主题的属性值，覆盖或补充公共配置：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class MultiTopicConsumers {

    @KafkaListener(
        topics = "topic1",
        properties = {
            "group-id=${topic1.consumer.group-id}",
            "key-deserializer=${topic1.consumer.key-deserializer}",
            "value-deserializer=${topic1.consumer.value-deserializer}",
            "spring.deserializer.key.delegate.class=${topic1.consumer.spring.deserializer.key.delegate.class}",
            "spring.deserializer.value.delegate.class=${topic1.consumer.spring.deserializer.value.delegate.class}"
        }
    )
    public void consumeTopic1(ConsumerRecord<String, Object> record) {
        // 处理topic1的消息逻辑
    }

    @KafkaListener(
        topics = "topic2",
        properties = {
            "group-id=${topic2.consumer.group-id}",
            "key-deserializer=${topic2.consumer.key-deserializer}",
            "value-deserializer=${topic2.consumer.value-deserializer}",
            "spring.deserializer.key.delegate.class=${topic2.consumer.spring.deserializer.key.delegate.class}",
            "spring.deserializer.value.delegate.class=${topic2.consumer.spring.deserializer.value.delegate.class}"
        }
    )
    public void consumeTopic2(ConsumerRecord<String, Object> record) {
        // 处理topic2的消息逻辑
    }
}

原理说明 #

• @KafkaListener的properties属性会直接覆盖Spring Boot自动配置的消费者属性，优先级更高。
• 公共配置放在spring.kafka.consumer下可以减少重复代码，只有每个主题独有的属性才需要在专属前缀下配置。
• 这种方式完全不需要自定义ContainerFactory，靠Spring原生的注解和配置机制就能实现多主题的独立消费者配置。

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