SpringBoot Kafka：“KafkaAutoConfiguration”中的 Bean 方法“kafkaTemplate”未加载

Question

我正在使用 springboot 并尝试编写一个 KafkaProducer 来将消息推送到 Kafka 队列中。 我在@Configuration 类中创建了这些方法。

@Bean
public KafkaTemplate<String, String> kafkaTemplate(){
    return new KafkaTemplate<>(producerFactory());
}

@Bean
public ProducerFactory<String, String> producerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress); //bootstrapAddress holds address of kafka server
    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    return new DefaultKafkaProducerFactory<>(configProps);
}

我已经在我的 KafkaMessageProducer 类中自动装配了这个 KafkaTemplate bean，它负责处理 KafkaTemplate 的发送功能。

@Autowired
KafkaTemplate<String, String> kafkaTemplate;

但是当我尝试编译我的代码时遇到了这个错误

Field kafkaTemplate in <pathoffile>.KafkaMessageProducer required a bean of type 'org.springframework.kafka.core.KafkaTemplate' that could not be found.
    - Bean method 'kafkaTemplate' in 'KafkaAutoConfiguration' not loaded because @ConditionalOnMissingBean (types: org.springframework.kafka.core.KafkaTemplate; SearchStrategy: all) found bean 'avroKafkaTemplate'
Action:Consider revisiting the conditions above or defining a bean of type 'org.springframework.kafka.core.KafkaTemplate' in your configuration.

另外，如果我尝试在我的 Spring 项目中排除 KafkaAutoConfiguration，我会收到类似“无法加载 Bean，因为 KafkaAutoConfiguration 已禁用”的错误。 知道为什么我会收到此 Bean 错误以及可能的解决方案吗？

编辑：- 我在我的项目使用的 jar 文件中发现了以下 bean

@Bean
@Conditional({EnableQueueCondition.class})
public KafkaTemplate<String, String> kafkaTemplate() {
    KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate(this.producerFactory());
    kafkaTemplate.setProducerListener(new ProducerListenerImpl());
    return kafkaTemplate;
}

所以，这就是错误的来源，但我不知道如何告诉 spring 不要查看这个 bean，并使用我定义的 bean。我尝试在 bean 上使用 Primary 和 Qualifier 注释，它仍然给出相同的错误。是否有可能未创建或找不到我定义的 bean，然后 KafkaAutoConfiguration 正在寻找被 avroKafkaTemplate bean 覆盖的默认 bean？有什么办法可以解决这个问题？

Answer 1

从堆栈跟踪中，还有另一个 KafkaTemplate bean - avroKafkaTemplate。所以我猜还有另一种配置，复制KafkaTemplate的定义。

Answer 2

如果你在 POM 中添加 kafka 依赖，spring boot 默认会提供 KafkaTemplate bean。

您只需要在 application.yml 文件中定义属性 示例：

server: port: 9000
spring:
   kafka:
     consumer:
        bootstrap-servers: localhost:9092
        group-id: group_id
        auto-offset-reset: earliest
        key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
        value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
     producer:
        bootstrap-servers: localhost:9092
        key-serializer: org.apache.kafka.common.serialization.StringSerializer
        value-serializer: org.apache.kafka.common.serialization.StringSerializer

启用自动配置

@Configuration
@EnableKafka

并自动装配 kafkaTemplate：

 @Autowired
 private KafkaTemplate<String, String> kafkaTemplate;

如果您的情况，自动配置工厂正在寻找与您的配置不匹配的ProducerFactory<String, Strng>。

 @Bean
 @ConditionalOnMissingBean(ProducerFactory.class)
        public ProducerFactory<String, Strng> kafkaProducerFactory()

所以将你的 producerFactory() 重命名为 kafkaProducerFactory()，它将解决你的问题。