函数式编程模型bean定义与spring-cloud-function + spring-cloud-stream集成

Question

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

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

例如，我有两个依赖项的 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 发布消息也是一样的想法，但是我如何从一个管道传输到另一个

所以我的问题与 according to spring documentation which says that I can use spring-cloud-stream as well:Wrappers for @Beans 类型的 Function、Consumer 和 Supplier 相关，将它们作为 HTTP 端点和/或带有 RabbitMQ、Kafka 等的消息流侦听器/发布者向外界公开，但我看不懂怎么办？

目前，不幸的是，我只能使用 Source see example here 手动将消息发布到 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 主体配置所需的管道链功能

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

干杯！

---

问候， 马克西姆

Answer 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();
    }
}

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

让我们先看看活页夹。

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

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

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

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

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

Answer 2

这是 Oleg Zhurakousky 的一个问题。如果能回答会很高兴

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

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

我只写生产者，它将向 Kafka 写入自定义 Pojo，而另一个应用程序是消费者。对于 Consumer<Pojo>...，函数式方法更清晰，它只会从 Kafka 中读取并处理。生产者的Supplier<Pojo>... 部分不清楚。

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

Answer 3

@阿布舍克

您可以按照here 的说明使用EmitterProcessor。提供的示例使用 rest 端点作为他的实际数据源，但正如您所见，它不会计量，因为您需要做的就是调用 EmitterProcessor 的 onNext 操作，从 Service 传递您的事件。