如何更正“spring 集成是单向 'MessageHandler' 并且不适合配置 'outputChannel'”异常？

Question

我查看了Spring Integration and DSL upgrade - one-way 'MessageHandler' and it isn't appropriate to configure 'outputChannel' Error，在我看来，我的解决方案的形状就在这里，但解决方案的表达方式对我来说没有意义。

spring-integration parallel split-route-aggregate flow fails due to one-way MessageHandler 更容易理解，但我没有看到处理程序方法（无论它是什么）在哪里返回一个 void 来给我这个打嗝。

为 .aggregate() 工厂方法抛出异常 spring integration is a one-way 'MessageHandler' and it isn't appropriate to configure 'outputChannel'。队列是在运行时根据应用程序收集的元数据动态构建的。

根据对框架的调试，.routeToRecipients(r -> this.buildRecipientListRouterSpecForRules(r, rules)) 子句似乎返回了一个 void。将.defaultOutputToParentFlow() 添加到 recipientListRouter 会消除异常，但可能不是正确的解决方案，因为当我进行此调整时流程实际上并没有开始。

我欢迎任何建议。

代码片段：

        StandardIntegrationFlow flow = IntegrationFlows
            .from(setupAdapter,
                    c -> c.poller(Pollers.fixedRate(1000L, TimeUnit.MILLISECONDS).maxMessagesPerPoll(1)))
            .enrichHeaders(h -> h.headerExpression("xxx", "payload[0].get(\"xxx\")")
                    .headerExpression("yyy", "payload[0].get(\"yyy\")")
                    )
            .split(tableSplitter)
            .enrichHeaders(h -> h.headerExpression("aaa", "payload[0].get(\"aaa\")")
                    .headerExpression("bbb", "payload[0].get(\"bbb\")")
                    )
            .channel(c -> c.executor(stepTaskExecutor))
            .routeToRecipients(r -> this.buildRecipientListRouterSpecForRules(r, rules))
            .aggregate()
            .handle(cleanupAdapter).get();

    return flow;



private RecipientListRouterSpec buildRecipientListRouterSpecForRules(RecipientListRouterSpec recipientListSpec,
        Collection<RuleMetadata> rules) {
    rules.forEach(
            rule -> recipientListSpec.recipientFlow(getFilterExpression(rule), f -> createFlowDefForRule(f, rule)));

    return recipientListSpec;
}

编辑：根据下面的讨论修改了解决方案的代码：

        StandardIntegrationFlow flow = IntegrationFlows
            .from(setupAdapter,
                    c -> c.poller(Pollers.fixedRate(1000L, TimeUnit.MILLISECONDS).maxMessagesPerPoll(1)))
            .enrichHeaders(h -> h.headerExpression("xxx", "payload[0].get(\"xxx\")")
                    .headerExpression("yyy", "payload[0].get(\"yyy\")")
                    )
            .gateway(new DirectChannel())
            .split(tableSplitter)
            .enrichHeaders(h -> h.headerExpression("aaa", "payload[0].get(\"aaa\")")
                    .headerExpression("bbb", "payload[0].get(\"bbb\")")
                    )
            .channel(c -> c.executor(stepTaskExecutor))
            .routeToRecipients(r -> this.buildRecipientListRouterSpecForRules(r, rules).defaultOutputToParentFlow())
            .aggregate()
            .handle(cleanupAdapter).get();

Answer 1

routeToRecipients() 之后您无法继续流程。我的意思是您的.aggregate() 导致了该错误，因为RecipientListRouter 不是AbstractReplyProducingMessageHandler，而只是普通的AbstractMessageHandler，并且它仅根据为该路由器类型提供的映射知道在路由功能之后将消息发送到哪里。

请参阅defaultOutputToParentFlow()JavaDocs 了解更多信息：

/**
 * Make a default output mapping of the router to the parent flow.
 * Use the next, after router, parent flow {@link MessageChannel} as a
 * {@link AbstractMessageRouter#setDefaultOutputChannel(MessageChannel)} of this router.
 * @return the router spec.
 */
public S defaultOutputToParentFlow() {

因此，只有当它与收件人映射不匹配时，它才会继续您的流程。 ignoreSendFailures(true) 也必须设置，以使其不会因错误而失败，而是回退到此默认输出。