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

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

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

我见过错误“是 Spring 集成聚合器 DSL 的单向“MessageHandler” https://stackoverflow.com/questions/40725701/error-is-a-one-way-messagehandler-for-spring-integration-aggregator-dsl，但那里的解决方案不适用于此处，我没有登录 handle() 方法。 我还尝试将 .defaultOutputToParentFlow() 添加到映射定义中，因为咖啡馆示例使用了它，但这也没有什么区别。

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

你的问题在这里：

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

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

.handle(new MessageHandler() {

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

注意void返回类型。由于没有什么可以返回，因此也没有什么可以发送到下游 -is a one-way '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);
        }
})

实际上我的想法是说我的变体，例如：

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

UPDATE

工作代码：

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