由于单向 MessageHandler，spring-integration 并行拆分路由聚合流失败

Question

我想通过拆分并行处理项目列表，将每个项目路由到它们适当的网关并聚合结果。但是，我的应用程序没有启动，出现以下错误:

BeanCreationException: The 'currentComponent' ... is a one-way 'MessageHandler' 
and it isn't appropriate to configure 'outputChannel'. 
This is the end of the integration flow.

这是一个示例流定义，它说明了行为:

@Bean
public IntegrationFlow parallelSplitRouteAggregateFlow() {
    return IntegrationFlows
            .from(Http.inboundGateway("/trigger"))
            .handle(message -> Arrays.asList(1, 2, 3))
            .split()
            .channel(MessageChannels.executor(Executors.newCachedThreadPool()))
            .<Integer, Boolean>route(o -> o % 2 == 0, m -> m
                    .subFlowMapping(true, oddFlow())
                    .subFlowMapping(false, evenFlow()))
            .aggregate()
            .get();
}

@Bean
public IntegrationFlow oddFlow() {
    return flow -> flow.<Integer>handle((payload, headers) -> "odd");
}

@Bean
public IntegrationFlow evenFlow() {
    return flow -> flow.<Integer>handle((payload, headers) -> "even");
}

我看过Error 'is a one-way 'MessageHandler' for spring-integration aggregator DSL ，但那里的解决方案不适用于此处，我没有登录 handle() 方法。我还尝试将 .defaultOutputToParentFlow() 添加到 mappingDefinition，因为咖啡馆示例使用了它，但这也没有区别。

我应该提一下，这是带有 spring-boot 2.0.1 版本的 spring-integration 5.0.4。

Answer 1

你的问题在这里:

.handle(message -> Arrays.asList(1, 2, 3))

如果您使用内联实现，它看起来像:

.handle(new MessageHandler() {

        @Override
        public void handleMessage(Message<?> message) throws MessagingException {
            Arrays.asList(1, 2, 3);
        }
})

注意 void 返回类型。由于没有要返回的内容，因此也没有要向下游发送的内容 - 是单向“MessageHandler”。

要解决您的问题，您需要这样做:

 .handle((p, h) -> Arrays.asList(1, 2, 3))

相当于:

.handle(new GenericHandler<Object>() {

        @Override
        public Object handle(Object p, Map<String, Object> h) {
            return Arrays.asList(1, 2, 3);
        }
})

实际上我的 IDEA 是这样说的:

这给了我一些提示，表明我做错了什么。

更新

工作代码:

@Bean
public IntegrationFlow parallelSplitRouteAggregateFlow() {
    return IntegrationFlows
            .from(Http.inboundGateway("/trigger"))
            .handle((p, h) -> Arrays.asList(1, 2, 3))
            .split()
            .channel(MessageChannels.executor(Executors.newCachedThreadPool()))
            .<Integer, Boolean>route(o -> o % 2 == 0, m -> m
                    .subFlowMapping(true, sf -> sf.gateway(oddFlow()))
                    .subFlowMapping(false, sf -> sf.gateway(evenFlow())))
            .aggregate()
            .get();
}

@Bean
public IntegrationFlow oddFlow() {
    return flow -> flow.<Integer>handle((payload, headers) -> "odd");
}

@Bean
public IntegrationFlow evenFlow() {
    return flow -> flow.<Integer>handle((payload, headers) -> "even");
}