【问题标题】：How to cascade Spring Cloud Dataflow streams?如何级联 Spring Cloud Dataflow 流？
【发布时间】：2023-04-09 20:15:01
【问题描述】：

我知道我可以在 Streams DSL 中使用扇入和扇出表示法在同一个流中级联多个流修饰符，例如：

s3 > :data
ftp > :data
http > :data
:data > file

但是，假设我注册了 stream1 和 stream2，每个都有自己的 DSL，我怎么能级联两者，做类似的事情？

例如：

stream1 DSL：

time | transform | file

stream2 DSL：

file | filter | http

然后我想创建类似stream3的东西：

stream1 | stream2 | s3

意味着time 和file 都将被读取，结果将被写入file、http 和s3。是否可以？

我假设除了写到我指定的接收器之外，stream1 和 stream2 也会写回管道，所以我可以级联它们并从 stream3 调用它们。

编辑 - 我要解决的问题。

正如 cmets 中所问的那样，让我详细说明我想在这里解决什么样的问题。

我工作的组织有复杂的流程，1 个团队无法维护整个事情，有数据生产者团队负责从源生成数据，直到最终的数据消费者通常有很多团队推断数据、变换、归一化等。

stream1，就我而言，将由公司的 1 个团队维护，steam2 由其他团队维护，stream3 由我的平台维护。

虽然维护每个流的团队是独立的，但从技术上讲，我仍然希望在聚合不同流时从内存管道中受益。当然，我希望自己必须管理自己的问题，例如 - 团队 1 发布了一个新版本的 stream1，这将需要一个新版本的 stream2 - 这对我的用例来说很好 - 我的团队可以解决这个问题。

【问题讨论】：

您能否编辑帖子并阐明此要求背后的原因？是“重用”吗？如果是重用，您是否仍然关心管道中应用程序的独立扩展或滚动升级？如果没有，你通过链接获得什么？
刚刚做了，请告诉我是否还有需要澄清的部分
感谢您提供详细信息。

标签： spring-cloud-stream spring-cloud-dataflow

【解决方案1】：

有几种方法可以解决这个问题。

方案一： 将处理器+接收器的职责组合成一个“处理器”；如果这样做了，您将能够将摄取（或）摄取 + 处理（或）摄取 + 处理 + 写入作为流管道的单独片段，并能够在 DSL 或 GUI 中使用它们。当然，这里组合的缺点是无法独立交互和升级业务逻辑，这对于一些用例来说可能不是问题。换句话说，您不一定需要将它们作为单独的微服务 - 如果职责很薄，将它们组合在单个 App 中就可以了。

选项 2： 不要使用 SCDF。如前所述构建一组类似的应用程序并将它们编排为独立应用程序。您不必处理 SCDF 中的源、处理器或接收器合同，因此您可以灵活地以任何您希望的方式插入应用程序。这里的缺点是手动编排。用于监控的数据管道可见性是另一回事。请参阅示例here。如果您仍想在 SCDF 中使用它，尽管它可能并不理想，但有一种解决方法 - 请参阅 here。

选项 3： 您还可以使用 Spring Integration 构建集成流，并将其构建到 SCSt 应用程序中。这可以提供更精细的控制，但您必须像在 Option-2 中那样单独编排它们。这是example。

【讨论】：

只是要清楚一点-选项2中的“手动编排”是指我必须使用消息队列而不是使用来自SCDF的内存管道？或者有没有一种简单的方法可以使用 spring 集成“手动”创建内存管道？
Spring Cloud Stream 应用程序，无论是手动部署还是通过 SCDF 部署，它们都通过消息代理（如 Kafka、RabbitMQ 或其他）相互通信。如果您想将一系列操作链接起来并在单个 JVM 中运行它们；基本上你希望它们在同一个线程上，你可以通过 Spring Integration Flow 来实现，如示例中所示。至于“手动编排”，我的意思是从您必须使用脚本和/或 CI/CD 工具将它们部署到云平台的角度来看。如果使用 SCDF，它会提供开箱即用的功能，我尝试列出不同之处。

【解决方案2】：

使用路由器接收器连接流。 @sabby？

https://github.com/spring-cloud/spring-cloud-stream-modules/tree/master/router-sink

【讨论】：