基于apache-kafka-binder的springCould Steam函数模型

Question

这是这个问题的续篇。我可以使用“普通的”Apache Kafka绑定器与功能模型一起使用吗？到目前为止，使用基于注释的配置，我混合了这两种方法，spring-cloud-stream-binder-kafka用于简单的消费/生产，spring-cloud-stream-binder-kafka-streams用于在一个应用程序中进行高级流处理。

函数模型似乎只有streams绑定器才支持，如果我尝试混合使用这两种方法--基于简单用法的注释和基于流的函数，则不会注册流绑定。

spring.cloud:
        stream:
          function:
            definition: processStream
          bindings:
            processStream-in-0:
              destination:  my-topic
            simple-binding-in:
              destination: another-topic

public interface SimpleBinding {

    String INPUT = "simple-binding-in";

    @Input(INPUT)
    SubscribableChannel simpleIn();

}

@Component
public class SimpleListener {

    @StreamListener(SimpleBinding.INPUT)
    public void listen(@Payload SomeDto payload) {
    }
}

@Configuration
public class FunctionalStream {

    @Bean
    public Consumer<KStream<String>> processStream() {
        return eventStream -> eventStream.map()
    }
}

配置类上存在@enablebinding(SimpleBinding.class)。是否首选/支持将两者混合使用，或者我是否应该使用streams-binder，甚至用于简单的消息消费？

Answer 1

对于Kafka Binder，您可以并且绝对应该使用functional model，同时忘记StreamListener。这样它将与您的KStream函数模型保持一致。

spring.cloud:
        stream:
          function:
            definition: processStream
          bindings:
            processStream-in-0:
              destination:  my-topic
            listen-in-0:
              destination: another-topic

@Component
public class SimpleListener {

    @Bean
    public Consumer<SomeDto> listen() {
        return payload -> ...
    }
}

@Configuration
public class FunctionalStream {

    @Bean
    public Consumer<KStream<String>> processStream() {
        return eventStream -> eventStream.map()
    }
}