Spring 云流函数:通过 REST 调用调用函数<T,R> 并将其输出到 KAFKA 主题

Spring Cloud Stream Function : Invoke Function<T,R> via REST call and output it to a KAFKA Topic

我有简单的 @Bean(Java 8 个函数)映射到目的地 topic-out-in)。

@Bean
public Function<String, String> transform() {
    return payload -> payload.toUpperCase();
}

@Bean
public Consumer<String> receive() {
    return payload -> logger.info("Data received: " + payload);
}

.yml 配置:

spring:
  cloud:
    stream:
      function:
        definition: transform;receive
      bindings:
        transform-out-0:
          destination: myTopic
        receive-in-0:
          destination: myTopic

现在,我想通过 REST 调用来调用 transform 函数,以便它的输出转到 destination topic(即 transform-out-0 映射到 myTopic) 并被 consumer 从此目的地拾取(receive-in-0 映射到 myTopic)。基本上,每个 REST 调用都应该生成一个 KAFKA Producer 的新实例并关闭它。

请问我如何使用 spring-cloud-stream 实现此目的?

谢谢

安舒曼

您应该使用 StreamBridge 而不是 transform 函数。这是 Spring Cloud Stream 中动态目标的新推荐方法。 这是基本思想:

@Autowired
private StreamBridge streamBridge;

@RequestMapping
public void delegateToSupplier(@RequestBody String body) {
    streamBridge.send("transform-out-0", body);
}

然后通过配置提供此 属性 - spring.cloud.stream.source: transform

Spring Cloud Stream 将为您创建一个名为 transform-out-0 的输出绑定。每次调用 REST 端点时,通过 StreamBridge,您会将数据发送到目标主题。

有关详细信息,请参阅 this