如何将“Kafka Streams Binder”与“Functional Style”和DI结合使用？

Question

https://cloud.spring.io/spring-cloud-static/spring-cloud-stream-binder-kafka/3.0.0.M3/reference/html/spring-cloud-stream-binder-kafka.html#_programming_model显示了一个示例，其中可以使用属性spring.cloud.stream.bindings.process_in.destination设置输入主题。

现在我想使用依赖注入，例如。

@Bean
public java.util.function.Consumer<KStream<Object, String>> process(JavaMailSender mailSender) {...}

启动应用程序（基于Spring Boot）时，属性为Spring。云流动绑定。process\u in。忽略目标，而订阅输入主题“输入”。

编辑：这是Kotlin代码（不含导入）

梅勒。kt:

@Configuration
class Mailer {
    @Bean
    fun sendMail(/*mailSender: JavaMailSender*/) = Consumer<KStream<Any, Mail>> { input ->
        input.foreach { _, mail -> println("mail = $mail") }
    }
}

邮政kt:

data class Mail(var from: String = "", var to: String = "", var subject: String = "", var body: String = "")

pplication.kt：

@SpringBootApplication
class Application

fun main(args: Array<String>) {
    runApplication<Application>(*args) {
    }
}

应用yml：：

spring.cloud.stream:
  bindings.sendMail_in.destination: mail
  kafka.binder.configuration.listeners: PLAINTEXT://localhost:9092

Answer 1

活页夹中存在一些问题，无法正确autowire提供给函数/消费者bean的bean。不过，最新的快照解决了这些问题。请确保您使用的是最新的快照（3.0.0. BUILD-SNAPSHOT）。这是一个示例应用程序，适用于您提供的相同场景。