Kafka消费者Spring靴

Question

使用spring集成Kafka dsl，我想知道为什么监听器不能接收消息？但是同样的应用程序，如果我用KafkaListener注释的方法替换spring integration DSL，就能够很好地使用消息。DSL让我错过了什么？

不消耗的DSL代码：

@Configuration
@EnableKafka
class KafkaConfig {
    //consumer factory provided by Spring boot
    @Bean
    IntegrationFlow inboundKafkaEventFlow(ConsumerFactory consumerFactory) {
        IntegrationFlows
                .from(Kafka
                .messageDrivenChannelAdapter(consumerFactory, "kafkaTopic")
                .configureListenerContainer({ c -> c.groupId('kafka-consumer-staging') })
                .id("kafkaTopicListener").autoStartup(true)
        )

                .channel("logChannel")
                .get()
    }
}

@Component
class KafkaConsumer {
    @KafkaListener(topics = ['kafkaTopic'], groupId = 'kafka-consumer-staging')
    void inboundKafkaEvent(String message) {
        log.debug("message is {}", message)
    }
}

Answer 1

您错过了使用Spring集成的事实，但您还没有在应用程序中启用它。但是，您不需要对Kafka执行此操作，因为您不会使用@kafkalistener来使用它。因此，要启用Spring集成基础结构，您需要在@configuration类上添加@enableintegration:https://docs.Spring.io/spring-integration/docs/5.1.6.release/reference/html/#configuration-enable-integration