子流支持

部分if…​elsepublish-subscribe组件提供了使用子流指定其逻辑或映射的功能。 最简单的示例是.publishSubscribeChannel(),如下例所示:spring-doc.cadn.net.cn

@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定义,但我们希望你发现 logic composition 的 sub-flow 风格很有用。 我们发现它会导致更短(因此更具可读性)的代码。spring-doc.cadn.net.cn

从版本 5.3 开始,BroadcastCapableChannel-基于publishSubscribeChannel()提供 implementation 以在 broker 支持的消息通道上配置子流订阅者。 例如,我们现在可以将多个订阅者配置为Jms.publishSubscribeChannel():spring-doc.cadn.net.cn

@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-subscribesub-flow 组合提供.routeToRecipients()方法。spring-doc.cadn.net.cn

另一个例子是使用.discardFlow()而不是.discardChannel().filter()方法。spring-doc.cadn.net.cn

.route()值得特别关注。 请考虑以下示例:spring-doc.cadn.net.cn

@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()继续像在常规Routermapping 的 map,但.subFlowMapping()将该子流绑定到 Main Flow。 换句话说,任何路由器的子流在.route().spring-doc.cadn.net.cn

有时,您需要引用现有的IntegrationFlow @Bean.subFlowMapping(). 以下示例显示了如何执行此作:spring-doc.cadn.net.cn

@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");
}


在这种情况下,当您需要接收来自此类子流的回复并继续主流时,此IntegrationFlowbean 引用(或其输入通道)必须用.gateway()如前面的示例所示。 这oddFlow()reference 未包装到.gateway(). 因此,我们不希望此路由分支回复。 否则,您最终会得到类似于以下内容的异常:spring-doc.cadn.net.cn

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 时,框架会处理与子流的请求-回复交互,并且不需要网关。spring-doc.cadn.net.cn

子流可以嵌套到任何深度,但我们不建议这样做。 事实上,即使在路由器的情况下,在流中添加复杂的子流也会很快开始看起来像一盘意大利面,人类很难解析。spring-doc.cadn.net.cn

在 DSL 支持子流配置的情况下,当正在配置的组件通常需要一个通道,并且该子流以channel()元素中,框架会隐式地放置一个bridge()在 Component Output Channel 和 Flow 的 Input Channel 之间。 例如,在这个filter定义:spring-doc.cadn.net.cn

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

框架在内部创建一个DirectChannelbean 中注入到MessageFilter.discardChannel. 然后,它将子流包装到IntegrationFlow从订阅的这个隐式通道开始,并放置一个bridgechannel()在流中指定。 当现有的IntegrationFlowbean 用作子流引用(而不是内联子流,例如 lambda),则不需要这样的桥,因为框架可以解析流 bean 中的第一个通道。 对于内联子流,输入通道尚不可用。spring-doc.cadn.net.cn