Spring Cloud Stream Kafka活页夹中的1个输出绑定

Question

在本页中，您不能只进行一次输出

但我只需要使用Spring Cloud Stream Kafka活页夹进行一次输出

那我该怎么办？

一些文章说使用org.springframework.cloud.stream.function.StreamBridge但它不适合我

我让StreamBridge向Kafka发送主题，但Kafka不会为我的Spring Boot应用程序生成主题

这是我的application.yml和生产主题代码

// producer Springboot application  



  spring.cloud.stream:
    kafka:
        binder:
            brokers: {AWS.IP}:9092
            zkNodes: {AWS.IP}:2181
    bindings:
        deliveryIncoming:
            destination : deliveryIncomingtopic

@Bean
public CommandLineRunner commandLineRunner(ApplicationContext ctx) {
  // wanna produce deliveryIncomingtopic and send to Spring's Consumer
}

// Consumer Springboot application

spring : 
     cloud:
        stream:
            kafka:
                binder: 
                    brokers: {AWS.IP}:9092
                    zkNodes: {AWS.IP}:2181
            function:
                definition : deliveryIncoming;

            bindings:
                deliveryIncoming-in-0:
                    destination : deliveryIncomingtopic

    @Bean
    public Consumer<KStream<String, String>> deliveryIncoming() { 
        return input ->
                input.foreach((key, value) -> {
                    System.out.println("deliveryIncoming is playing");
                    System.out.println("Key: " + key + " Value: " + value);
                }); 
    }

对不起，我想我说的有点不清楚

我只想像下面这样做

生产（deliveryIncomingtopic）-

Answer 1

我只是想把事情弄清楚。您是否希望使用入站主题中的消息，然后生成/生成另一条消息到另一个输出主题？如果这是您的问题，那么我相信您在应用程序中遗漏了一些东西。yml。您需要为输出主题进行其他配置。例如：

由于您使用的是Spring Cloud Function（基于我在您的application.yml中看到的内容），因此您应该为输出主题添加更多配置，如下所示：

spring : 
     cloud:
        stream:
            kafka:
                binder: 
                    brokers: {AWS.IP}:9092
                    zkNodes: {AWS.IP}:2181
            function:
                definition: deliveryIncoming;deliveryOutput

            bindings:
                deliveryIncoming-in-0:
                    destination: deliveryIncomingtopic
                deliveryOutput-out-0:
                    destination: deliveryOutput

并为producer函数定义另一个bean：

@Bean
public Producer<KStream<String, String>> deliveryOutput() { 
   // do your necessary logic here to outbound your message
}

希望这能符合你的期望。

Answer 2

如果是这种情况，那么您需要更改bean函数定义以返回java.util.Function而不是java.util.消费者。

@Bean
public Function<KStream<String, String>, KStream<String, String>> deliveryIncoming() {
return input ->
                input.foreach((key, value) -> {
                    System.out.println("deliveryIncoming is playing");
                    System.out.println("Key: " + key + " Value: " + value);
                }); 
}

但是，AFAIK...您仍然需要在application.yml.中定义输出通道。您可以使用相同的名称和不同的后缀。如下所示：

deliveryIncoming-in-0:
   destination: <your_topic_name>
deliveryIncoming-out-0:
   destination: <your_topic_name>