子流支持

某些 if…​elsepublish-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 版本开始,提供了基于 BroadcastCapableChannelpublishSubscribeChannel() 实现,用于在代理支持的消息通道上配置子流订阅者。例如,我们现在可以在 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 支持子流配置的情况下,当通常需要通道用于正在配置的组件时,并且该子流以 channel() 元素开头时,框架会在组件输出通道和流程输入通道之间隐式放置一个 bridge()。例如,在这个 filter 定义中

.filter(p -> p instanceof String, e -> e
	.discardFlow(df -> df
                         .channel(MessageChannels.queue())
                         ...)

框架在内部创建一个 DirectChannel bean 以注入到 MessageFilter.discardChannel 中。然后,它将子流封装到一个以这个隐式通道开头的 IntegrationFlow 中进行订阅,并在流程中指定的 channel() 之前放置一个 bridge。当使用现有的 IntegrationFlow bean 作为子流引用(而不是内联子流,例如 lambda)时,不需要这样的桥接,因为框架可以从流程 bean 中解析第一个通道。使用内联子流时,输入通道尚不可用。

© . This site is unofficial and not affiliated with VMware.