Integration Flows Composition

随着 MessageChannel 抽象在 Spring Integration 中作为一等公民,被假定总是构成集成流。流中任何端点的输入通道可用于从任何其他端点发送消息,而不仅仅是将此通道作为输出的端点。此外,借助 @MessagingGateway 约定、内容扩充组件、复合端点(例如 <chain>),以及现在借助 IntegrationFlow bean(例如 IntegrationFlowAdapter),在更短、可重用部分之间分配业务逻辑非常直接。最终构成所需的所有内容都是要发送到或从中接收的 MessageChannel 的知识。

With the MessageChannel abstraction as a first class citizen in Spring Integration, the composition of integration flows was always assumed. The input channel of any endpoint in the flow can be used to send messages from any other endpoint and not only from the one which has this channel as an output. Furthermore, with a @MessagingGateway contract, Content Enricher components, composite endpoints like a <chain>, and now with IntegrationFlow beans (e.g. IntegrationFlowAdapter), it is straightforward enough to distribute the business logic between shorter, reusable parts. All that is needed for the final composition is knowledge about a MessageChannel to send to or receive from.

5.5.4 版开始,为了从 MessageChannel 中抽象更多内容并向最终用户隐藏实现细节,IntegrationFlow 引入了 from(IntegrationFlow) 工厂方法,以允许从现有流的输出启动当前 IntegrationFlow

Starting with version 5.5.4, to abstract more from MessageChannel and hide implementation details from the end-user, the IntegrationFlow introduces the from(IntegrationFlow) factory method to allow starting the current IntegrationFlow from the output of an existing flow:

@Bean
IntegrationFlow templateSourceFlow() {
    return IntegrationFlow.fromSupplier(() -> "test data")
            .channel("sourceChannel")
            .get();
}

@Bean
IntegrationFlow compositionMainFlow(IntegrationFlow templateSourceFlow) {
    return IntegrationFlow.from(templateSourceFlow)
            .<String, String>transform(String::toUpperCase)
            .channel(c -> c.queue("compositionMainFlowResult"))
            .get();
}

另一方面,IntegrationFlowDefinition 添加了一个 to(IntegrationFlow) 终端运算符,以在其他流的输入通道继续当前流:

On the other hand, the IntegrationFlowDefinition has added a to(IntegrationFlow) terminal operator to continue the current flow at the input channel of some other flow:

@Bean
IntegrationFlow mainFlow(IntegrationFlow otherFlow) {
    return f -> f
            .<String, String>transform(String::toUpperCase)
            .to(otherFlow);
}

@Bean
IntegrationFlow otherFlow() {
    return f -> f
            .<String, String>transform(p -> p + " from other flow")
            .channel(c -> c.queue("otherFlowResultChannel"));
}

流中间的构成只需要一个现有的 gateway(IntegrationFlow) EIP 方法即可实现。这样,我们可以通过从更简单、可重用的逻辑块构成流来构建任意复杂程度的流。例如,你可以将一个 IntegrationFlow bean 库添加为依赖项,只需将它们的配置类导入到最终项目并针对你的 IntegrationFlow 定义自动装配。

The composition in the middle of the flow is simply achievable with an existing gateway(IntegrationFlow) EIP-method. This way we can build flows with any complexity by composing them from simpler, reusable logical blocks. For example, you may add a library of IntegrationFlow beans as a dependency, and it is just enough to have their configuration classes imported to the final project and autowired for your IntegrationFlow definitions.