如何从application.yaml加载Kafka消费者/生产者配置属性

1. 无需手动编写配置项，从application.yaml加载配置创建DefaultKafkaConsumerFactory #

Spring Boot会自动将配置文件中spring.kafka.*前缀的所有Kafka相关配置绑定到KafkaProperties内置Bean中，直接注入该Bean即可拿到完整的消费者配置，不需要手动编写consumerProps()方法逐行赋值。
首先确保配置文件中的Kafka配置使用标准前缀，示例配置如下：

spring:
  kafka:
    bootstrap-servers: 你的Kafka服务地址:9092
    consumer:
      group-id: 消费组ID
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "*"
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

2. 修复java.lang.IllegalStateException: a KafkaTemplate is required to support replies异常 #

异常触发原因：

• 调整代码时，你删除了原先手动定义ConcurrentKafkaListenerContainerFactory的逻辑，Spring Boot自动创建的默认Listener工厂未配置replyTemplate属性，不支持请求-回复的消息模式。
• 原有代码中factory.setReplyTemplate(kafkaTemplate())的配置逻辑丢失。

修复后可直接运行的KafkaConfig代码 #

@Configuration
@EnableKafka
public class KafkaConfig {

    private final KafkaProperties kafkaProperties;

    // 构造注入自动绑定的Kafka配置对象
    public KafkaConfig(KafkaProperties kafkaProperties) {
        this.kafkaProperties = kafkaProperties;
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<Object, KafkaExampleRecord> kafkaListenerContainerFactory(
            KafkaTemplate<Object, KafkaExampleRecord> kafkaTemplate) {
        ConcurrentKafkaListenerContainerFactory<Object, KafkaExampleRecord> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        // 直接从KafkaProperties获取解析好的消费者配置，无需手动写consumerProps()
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(kafkaProperties.buildConsumerProperties()));
        // 显式注入回复用的KafkaTemplate，解决启动异常
        factory.setReplyTemplate(kafkaTemplate);
        return factory;
    }

    @Bean
    public ReplyingKafkaTemplate<Object, KafkaExampleRecord, KafkaExampleRecord> replyingKafkaTemplate(
            ProducerFactory<Object, KafkaExampleRecord> producerFactory,
            ConcurrentMessageListenerContainer<Object, KafkaExampleRecord> repliesContainer) {
        return new ReplyingKafkaTemplate<>(producerFactory, repliesContainer);
    }

    @Bean
    public ConcurrentMessageListenerContainer<Object, KafkaExampleRecord> repliesContainer(
            ConcurrentKafkaListenerContainerFactory<Object, KafkaExampleRecord> kafkaListenerContainerFactory) {
        ConcurrentMessageListenerContainer<Object, KafkaExampleRecord> rc =
                kafkaListenerContainerFactory.createContainer("mytopic");
        rc.setAutoStartup(false);
        return rc;
    }

    @Bean
    public KafkaTemplate<Object, KafkaExampleRecord> kafkaTemplate(ProducerFactory<Object, KafkaExampleRecord> pf) {
        return new KafkaTemplate<>(pf);
    }
}

注意事项 #

• 所有自定义Kafka参数（包括超时时间、拦截器、序列化配置等）都可以直接写在application.yaml对应前缀路径下，KafkaProperties会自动识别加载，不需要在Java代码中手动set。
• 不要在@Bean方法中直接调用其他@Bean方法获取实例，统一通过方法参数注入依赖，避免代理场景下出现多实例、依赖顺序错误的问题。

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