spring-boot - Spring Boot 上 Spring 集成的 Kafka 配置

Question

我有一个工作原型(prototype) Spring Boot 应用程序，它监听 Kafka 队列。除了application.yml中的配置，只需要一个 MessageListener用 @KafkaListener 注释的实现.

我现在介绍 Spring Integration，并为此配置了这些 bean:

@Bean
public KafkaMessageDrivenChannelAdapter<String, String>
adapter(KafkaMessageListenerContainer<String, String> container) {
    KafkaMessageDrivenChannelAdapter<String, String> kafkaMessageDrivenChannelAdapter =
            new KafkaMessageDrivenChannelAdapter<>(container);
    kafkaMessageDrivenChannelAdapter.setOutputChannel(receiver());
    return kafkaMessageDrivenChannelAdapter;
}

@Bean
public KafkaMessageListenerContainer<String, String> container() throws Exception {
    ContainerProperties properties = new ContainerProperties(this.topic);
    // set more properties
    return new KafkaMessageListenerContainer<>(consumerFactory(), properties);
}

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    Map<String, Object> props = ...; // set proerties
    return new DefaultKafkaConsumerFactory<>(props);
}

应用程序未启动，并引发此错误:

Parameter 1 of method kafkaListenerContainerFactory in org.springframework.boot.autoconfigure.kafka.KafkaAnnotationDrivenConfiguration required a bean of type 'org.springframework.kafka.core.ConsumerFactory' that could not be found.
    - Bean method 'kafkaConsumerFactory' in 'KafkaAutoConfiguration' not loaded because @ConditionalOnMissingBean (types: org.springframework.kafka.core.ConsumerFactory; SearchStrategy: all) found bean 'consumerFactory'


Action:

Consider revisiting the conditions above or defining a bean of type 'org.springframework.kafka.core.ConsumerFactory' in your configuration.

即使我已经定义了 ConsumerFactory bean 。

在 Debug模式下运行，很明显 Boot 正在加载 KafkaListenerEndpointContainer bean 监听代理:

[org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : Discovered coordinator localhost:9092 (id: 2147483999 rack: null) for group my_group.

然后:

  KafkaAnnotationDrivenConfiguration matched:
      - @ConditionalOnClass found required class 'org.springframework.kafka.annotation.EnableKafka'; @ConditionalOnMissingClass did not find unwanted class (OnClassCondition)

   KafkaAnnotationDrivenConfiguration#kafkaListenerContainerFactory matched:
      - @ConditionalOnMissingBean (names: kafkaListenerContainerFactory; SearchStrategy: all) did not find any beans (OnBeanCondition)

   KafkaAnnotationDrivenConfiguration#kafkaListenerContainerFactoryConfigurer matched:
      - @ConditionalOnMissingBean (types: org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer; SearchStrategy: all) did not find any beans (OnBeanCondition)

   KafkaAnnotationDrivenConfiguration.EnableKafkaConfiguration matched:
      - @ConditionalOnMissingBean (names: org.springframework.kafka.config.internalKafkaListenerAnnotationProcessor; SearchStrategy: all) did not find any beans (OnBeanCondition)

   KafkaAutoConfiguration matched:
      - @ConditionalOnClass found required class 'org.springframework.kafka.core.KafkaTemplate'; @ConditionalOnMissingClass did not find unwanted class (OnClassCondition)

   KafkaAutoConfiguration#kafkaProducerFactory matched:
      - @ConditionalOnMissingBean (types: org.springframework.kafka.core.ProducerFactory; SearchStrategy: all) did not find any beans (OnBeanCondition)

   KafkaAutoConfiguration#kafkaProducerListener matched:
      - @ConditionalOnMissingBean (types: org.springframework.kafka.support.ProducerListener; SearchStrategy: all) did not find any beans (OnBeanCondition)

   KafkaAutoConfiguration#kafkaTemplate matched:
      - @ConditionalOnMissingBean (types: org.springframework.kafka.core.KafkaTemplate; SearchStrategy: all) did not find any beans (OnBeanCondition)

我认为正在发生的事情是 Spring Boot\Kafa 自动配置与 Spring Integration\Kafka 设置发生冲突。解决此问题的正确方法是什么？

谢谢

Answer 1

您可以使用 Boot 的消费者工厂...

@Bean
public KafkaMessageListenerContainer<String, String> container(ConsumerFactory<String, String> cf) {
    ...
}

或者禁用 kafka 自动配置

@SpringBootApplication(exclude = KafkaAutoConfiguration.class)