如何纠正“Spring integration是单向的'MessageHandler'，不适合配置'Output Channel'”异常？

Question

我已经查看了Spring Integration and DSL Upgrade-One-way“MessageHandler”，配置“Output Channel”错误是不合适的，在我看来，我的解决方案的形状就在这里，但解决方案的表达方式对我来说是没有意义的。

spring-integration并行分裂路由聚合流失败，原因是单向MessageHandler更容易理解，但我没有看到处理程序方法（无论它是什么）在哪里返回了一个空白，从而导致了这个打嗝。

异常Spring integration是一个单向的“MessageHandler”，不适合为.aggregate（）工厂方法设置“output channel”。队列是基于应用程序收集的元数据在运行时动态构造的。

根据对框架的调试，看起来.routeToreCipients(r->this.buildrecipientlistrouterspecforrules(r,rules））子句返回一个void。向recipientListRouter添加.DefaultOutputToparentFlow（）可以消除异常，但可能不是正确的解决方案，因为当我进行此调整时，流实际上并没有启动。

我欢迎任何建议。

代码片段：

        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()之后继续该流程。我的意思是您的.Aggrege()导致了该错误，因为RecipientListRouter不是AbstractReplyProductingMessageHandler，而只是普通的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() {

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