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

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'是spring-integration aggregator DSL的单向'MessageHandler'，但是这里的解决方案不适用，我没有登录handle（）方法。我还试图将。DefaultOutputToparentFlow（）添加到mappingDefinition中，因为cafe示例使用它，但这也没有什么不同。

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

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