0

大家好，特别是春季团队！

如何以函数式 Bean 编程模型风格使用 spring-cloud-stream 管道 spring-cloud-function？

例如，我有两个依赖项的 pom.xml：

<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-stream-reactive</artifactId>
</dependency>
<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-starter-function-webflux</artifactId>
</dependency>

假设我接下来想做：

  1. 通过 spring-cloud-function (webflux) 通过 http 有效负载字符串发送
  2. 使用我的 toUpperCase 函数将其大写
  3. 最后发送到我的管道中以安装 binder (kafka/rabbit/test-binder)

所以我期待像这样实现它：

@Log4j2
@SpringBootApplication
public class SpringCloudFunctionStreamApplication {

  /**
   * can I sent result of that function to my broker without any
   * explicitly defined output.send(...) execution?
   */
  @Bean
  public Function<String, String> toUpperCase() {
    return arg -> {
      var res = arg.toUpperCase();
      log.info("toUpperCase: {}", res);
      return res;
    };
  }

  public static void main(String[] args) {
    SpringApplication.run(
      SpringCloudFunctionStreamApplication.class,
      "--spring.cloud.function.definition=toUpperCase",
      "--spring.cloud.stream.function.definition=toUpperCase"
    );
  }
}

因此，当我使用 HTTPie 发送有效负载时，如下所示：

echo 'hello' | http :8080/toUpperCase

spring-cloud-function 似乎工作正常，我可以看到预期的日志：

2019-06-09 21:20:36.978 ...SpringCloudFunctionStreamApplication : toUpperCase: hello

同样的想法，如果我通过 rabbitmq 管理 web ui 发布消息，但是我如何从一个管道到另一个管道

所以我的问题与根据spring文档相关，该文档说我也可以使用spring-cloud-stream： Wrappers for @Beans of type Function, Consumer and Supplier，将它们作为HTTP端点和/或消息流暴露给外界使用 RabbitMQ、Kafka 等的听众/发布者，但我不明白怎么做？

目前，不幸的是，我只能使用 Source 手动将消息发布到 spring-cloud-stream binder，请参见此处的示例，但我当然想知道是否可以使用 spring 来避免，神奇地......

请任何人告诉我（可能是 Gary Russell、Dave Sawyer、Artem Bilan、Oleg Zhurakousky 或其他任何知道的人）：我错过了什么以及我应该如何配置我的应用程序或者我应该在我的 application.properties 中添加哪些道具等？

谢谢！

更新

已经有一段时间了，但我决定在这里发布一个解决方案......

简而言之：固定提交在这里：https ://github.com/daggerok/spring-cloud-function-stream-integration/commit/35325465b81bb869c31ec7892f413ab891d6d0fd

所以基本上使用 StreamBridge 我可以将 spring-cloud-function 与 spring-cloud-stream 连接起来......我可以在适当的目的地发送任何想要的东西，可以直接从我的 spring-cloud-function 主体配置所需的管道链

在提到的更新存储库中查看详细信息（如果您愿意，可以修复分支）

干杯!

问候，马克西姆

4

3 回答 3

1

马克西姆

这并不理想，但鉴于我们在现有绑定器的范围内实现了初始功能支持，因此存在一些限制。我会解释，但首先这里是功能齐全的代码：

@SpringBootApplication
public class SimpleFunctionRabbitDemoApplication  {

    public static void main(String[] args) throws Exception {
        SpringApplication.run(SimpleFunctionRabbitDemoApplication.class,
            "--spring.cloud.stream.function.definition=uppercase");
    }

    @Autowired
    private Processor processor;

    @Bean
    public Consumer<String> consume() {
        return v -> processor.input().send(MessageBuilder.withPayload(v).build());
    }

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

基本上有一点不匹配。在流方面，我们有活页夹，在功能方面，我们有适配器。您正在有效地（根据您的要求）尝试将两者连接到管道中。所以。. .

我们先来看看绑定器。

大写函数绑定到input消息output通道绑定器（rabbit 或 kafka）提供的通道，有效地创建了内部管道input -> uppercase -> output。它也被 sc-function 公开为 REST 端点，但是，sc-function 无权访问上述管道。事实上它实际上有它自己的管道request -> uppercase -> reply。所以我们需要做的是将这两个概念连接在一起，这实际上就是我所做的。

  • 您使用绑定注入您的应用程序，Processor其中包含对uppercase绑定到的通道的引用。

  • consume()您通过 REST调用http://localhost:8080/consume/blah。

  • uppercase您向函数的输入通道发送消息

将来为了简化这一点，我们只需要创建一个类似 binder 的 Web 适配器版本，因此请随时提出功能请求。但正如您所看到的，当前的解决方法并不是全部。

于 2019-06-11T17:48:51.593 回答
0

这是奥列格·朱拉库斯基的一个问题。如果回答会很高兴

如果我用于绑定输出目标，每次将新的发送到 Kafka/Rabbit时@Bean Supplier<Pojo>...如何从一个@Service类或类调用它。@ControllerPojo

Supplier只公开一个get()方法。

我只写生产者，它将Pojo为 Kafka 编写自定义，而另一个应用程序是消费者。Consumer<Pojo>...对于只从 Kafka 和进程中读取的情况，函数式方法更加清晰。制作人的Supplier<Pojo>...部分不清楚。

https://www.youtube.com/watch?v=nui3hXzcbK0&t=3478s

于 2019-07-01T10:46:22.687 回答
0

@Abhishek

您可以EmitterProcessor按照此处所述使用。提供的示例使用休息端点作为他的实际数据源，但正如您所见，它没有计量，因为您需要做的就是调用从服务传递事件的onNext操作。EmitterProcessor

于 2020-02-04T05:37:37.260 回答