子流支持
一些 if…else
和 publish-subscribe
组件提供了使用子流来指定其逻辑或映射的能力。
最简单的示例是 .publishSubscribeChannel()
,如下例所示:
@Bean
public IntegrationFlow subscribersFlow() {
return flow -> flow
.publishSubscribeChannel(Executors.newCachedThreadPool(), s -> s
.subscribe(f -> f
.<Integer>handle((p, h) -> p / 2)
.channel(c -> c.queue("subscriber1Results")))
.subscribe(f -> f
.<Integer>handle((p, h) -> p * 2)
.channel(c -> c.queue("subscriber2Results"))))
.<Integer>handle((p, h) -> p * 3)
.channel(c -> c.queue("subscriber3Results"));
}
您可以通过单独的 IntegrationFlow
@Bean
定义实现相同的结果,但我们希望您发现子流风格的逻辑组合很有用。
我们发现它能使代码更短(因此更具可读性)。
从 5.3 版本开始,publishSubscribeChannel()
提供了一个基于 BroadcastCapableChannel
的实现,用于在 broker 支持的消息通道上配置子流订阅者。
例如,我们现在可以在 Jms.publishSubscribeChannel()
上将多个订阅者配置为子流:
@Bean
public JmsPublishSubscribeMessageChannelSpec jmsPublishSubscribeChannel() {
return Jms.publishSubscribeChannel(jmsConnectionFactory())
.destination("pubsub");
}
@Bean
public IntegrationFlow pubSubFlow(BroadcastCapableChannel jmsPublishSubscribeChannel) {
return f -> f
.publishSubscribeChannel(jmsPublishSubscribeChannel,
pubsub -> pubsub
.subscribe(subFlow -> subFlow
.channel(c -> c.queue("jmsPubSubBridgeChannel1")))
.subscribe(subFlow -> subFlow
.channel(c -> c.queue("jmsPubSubBridgeChannel2"))));
}
类似的 publish-subscribe
子流组合提供了 .routeToRecipients()
方法。
另一个示例是在 .filter()
方法上使用 .discardFlow()
而不是 .discardChannel()
。
.route()
值得特别关注。
考虑以下示例:
@Bean
public IntegrationFlow routeFlow() {
return f -> f
.<Integer, Boolean>route(p -> p % 2 == 0,
m -> m.channelMapping("true", "evenChannel")
.subFlowMapping("false", sf ->
sf.<Integer>handle((p, h) -> p * 3)))
.transform(Object::toString)
.channel(c -> c.queue("oddChannel"));
}
.channelMapping()
继续像在常规 Router
映射中一样工作,但 .subFlowMapping()
将该子流绑定到主流。
换句话说,任何路由器的子流在 .route()
之后都会返回到主流程。
有时,您需要从 .subFlowMapping()
引用现有的 IntegrationFlow
@Bean
。
以下示例展示了如何实现:
@Bean
public IntegrationFlow splitRouteAggregate() {
return f -> f
.split()
.<Integer, Boolean>route(o -> o % 2 == 0,
m -> m
.subFlowMapping(true, oddFlow())
.subFlowMapping(false, sf -> sf.gateway(evenFlow())))
.aggregate();
}
@Bean
public IntegrationFlow oddFlow() {
return f -> f.handle(m -> System.out.println("odd"));
}
@Bean
public IntegrationFlow evenFlow() {
return f -> f.handle((p, h) -> "even");
}
在这种情况下,当您需要从这样的子流接收回复并继续主流程时,此 IntegrationFlow
bean 引用(或其输入通道)必须用 .gateway()
包装,如上例所示。
上例中的 oddFlow()
引用没有被包装到 .gateway()
中。
因此,我们不期望从这个路由分支得到回复。
否则,您会遇到类似于以下的异常:
Caused by: org.springframework.beans.factory.BeanCreationException:
The 'currentComponent' (org.springframework.integration.router.MethodInvokingRouter@7965a51c)
is a one-way 'MessageHandler' and it isn’t appropriate to configure 'outputChannel'.
This is the end of the integration flow.
当您将子流配置为 lambda 时,框架会处理与子流的请求-回复交互,并且不需要网关。
子流可以嵌套到任何深度,但我们不建议这样做。 事实上,即使在路由器的情况下,在流中添加复杂的子流也很快会变得像一盘意大利面条,让人难以理解。
在 DSL 支持子流配置的情况下,当通常需要为所配置的组件提供通道,并且该子流以
框架内部创建一个 |