基于apache-kafka-binder的Spring-cloud-stream功能模型

Question

这是这个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()
    }
}

configuration类中存在@EnableBinding(SimpleBinding.class)。是否更喜欢/支持按照描述将两者混合在一起，或者我是否应该使用streams-binder，甚至用于简单的消息消费？

Answer 1

对于Kafka Binder，你可以而且绝对应该使用功能模型，完全忘记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()
    }
}