Spring的云流Kafka发送消息

Question

如何使用新的Spring Cloud Stream Kafka功能模型发送消息？

不推荐的方式是这样的。

public interface OutputTopic {

    @Output("output")
    MessageChannel output();
}

@Autowired
OutputTopic outputTopic;

public someMethod() {
    
    outputTopic.output().send(MessageBuilder.build());
}

但是我如何以函数式风格发送消息呢？

应用yml公司

spring:
  cloud:
    function:
      definition: process
    stream:
      bindings:
        process-out-0:
          destination: output
          binder: kafka

@Configuration
public class Configuration {

    @Bean
    Supplier<Message<String>> process() {
        return () -> {
            return MessageBuilder.withPayload("foo")
                    .setHeader(KafkaHeaders.MESSAGE_KEY, "bar".getBytes()).build();
        };
    }

我会自动连接MessageChannel，但对于process、process-out-0、output或类似的东西，没有MessageChannel Bean。或者我可以用供应商Bean发送消息吗？谁能给我举个例子吗？谢谢！

Answer 1

您可以使用StreamBridge或反应器API-请参阅将任意数据发送到输出（例如外部事件驱动源）