报头聚合期间错误通道和回复通道消失

Question

我有以下工作流程。

1. 入站通道

2. 分离器

3. 拆分通道的任务执行器 - 所有线程执行相同的工作流。

3.a.构造请求

3.b。网关消息端点的服务激活器包装器。

3.c。使用错误通道配置的 http-outbound-gateway 上的网关包装器（在调用 http-outbound-gateway 时处理异常）

3.d。 http-outbound-gateway

4. 聚合器

5. 响应 Spring 集成工作流。

如果在 3.d 中发生异常，则控制权转到为网关错误通道配置的服务激活器。 我将以下内容从失败的消息复制到新标头，再复制到传递给错误通道的标头。

一个。相关 ID 湾。序列号 C。序列大小

但在聚合拆分器响应时，DefaultAggregatingMessageGroupProcessor.java 会删除冲突的标头，并在将控制权提供给聚合器之前删除错误通道和回复通道。

因此，一旦聚合器完成其操作，它就无法找到回复或错误通道，并导致异常。

我使用的是 spring-integration-core 版本 2.2.1，我不确定为什么在标头聚合期间会删除回复通道和错误通道。

任何有关解决此问题的意见都会有很大帮助。

谢谢你:)

编辑 1： 非常感谢 Gary 帮助我解决这个问题。我正在分享我当前的配置

<!-- SPLITTER -->
<int:splitter id="dentalSplitter" ref="dentalServiceSplitter"
    method="getDentalServiceList" input-channel="dentalServiceSplitterChannel"
    output-channel="dentalSplitterTaskChannel" />

<int:channel id="dentalSplitterTaskChannel">
    <int:dispatcher task-executor="dentalTaskExecutor" />
</int:channel>

<int:chain input-channel="dentalSplitterTaskChannel" output-channel="dentalGatewayChannel">
    <int:header-enricher>
        <int:header name="CHAIN_START_TIME" expression="T(System).currentTimeMillis()" overwrite="true" />
    <int:object-to-json-transformer content-type="application/json"/>               
</int:chain>

<int:service-activator input-channel="dentalGatewayChannel" ref="dentalGatewayWrapper" output-channel="dentalReplyChannel" />
<int:gateway id="dentalGatewayWrapper" default-request-channel="dentalCostEstimateRequestChannel" error-channel="dentalErrorChannel"/>

<int-http:outbound-gateway id="dentalGateway"
    url-expression="@urlBuilder.build('${service.endpoint}')"  
    http-method="POST" request-factory="clientHttpRequestFactory"  
    request-channel="dentalCostEstimateRequestChannel" extract-request-payload="true" 
    expected-response-type="com.dental.test.DentalResponse">
    <int-http:request-handler-advice-chain> 
         <ref bean="logChainTimeInterceptor" /> 
    </int-http:request-handler-advice-chain>
</int-http:outbound-gateway>

<!-- EXCEPTION -->
<int:chain input-channel="dentalErrorChannel" output-channel="dentalAggregatorChannel">
    <int:transformer ref="commonErrorTransformer" method="dentalGracefulReturn"/>
</int:chain>

<!-- SUCCESS -->
<int:chain input-channel="dentalReplyChannel" output-channel="dentalAggregatorChannel">
        <int:filter discard-channel="dentalErrorChannel"
        expression="T(com.dental.util.InvocationOutcomeHelper).isOutcomeSuccess(payload?.metadata?.outcome?.code,payload?.metadata?.outcome?.message)" />
</int:chain>

<!-- AGGREGATION -->

<int:chain input-channel="dentalAggregatorChannel" output-channel="wsDentalServiceOutputChannel" >
    <int:aggregator ref="dentalServiceAggregator" />
    <int:service-activator ref="dentalResponseServiceActivator" />
</int:chain>

我注意到的是，每个拆分通道在通过网关时都会为错误和回复创建一个新的临时通道，并且在从网关返回响应后，它会保留保留的（原始入站）错误和回复通道标头。正如您所提到的，在控制到达错误转换器之后，保留保留的标头的流程被破坏，并且聚合消息组处理器接收到三个不同的临时通道实例，因此将它们删除。 我打算有一个自定义消息组处理器并修改用于聚合标头的冲突解决策略并提出此配置。

<bean id="channelPreservingAggregatingMessageHandler" class="org.springframework.integration.aggregator.AggregatingMessageHandler">
    <constructor-arg name="processor" ref="channelPreservingMessageGroupProcessor"/>
</bean>

不过，我还没有对此进行测试。但根据这个讨论，这看起来不是一个可行的解决方案。

看起来我在网关中的错误处理配置不正确。 但是，我对您的这句话感到困惑“与其直接转发消息，不如简单地处理错误流中的错误并将结果正常返回给网关“包装器””。如果我删除错误通道，当异常发生时我将如何恢复控制？可能是我想在这里理解一些东西。请您详细说明一下吗？

Answer 1

在询问有关此类场景的问题时，您通常需要展示您的配置。但是，我怀疑您正在将错误流中的消息直接转发到聚合器。

这就像在代码中执行 GOTO 并破坏范围。

它不起作用，因为错误消息中的 replyChannel 标头用于网关“包装器”，而不是原始上游入站网关。当聚合器获得冲突的标头时，它别无选择，只能删除标头（您将看到一条关于该效果的 DEBUG 日志消息）。

不要直接转发消息，只需处理错误流上的错误并将结果正常返回到网关“包装器”（只需省略错误流上最后一个元素上的错误通道）。

网关随后将修复回复，使其与其他消息（好消息和坏消息）保持一致，并将其转发给聚合器。

您不需要在错误流中弄乱标题，只需返回您想要聚合的值以及良好的结果。

您真的应该更新到current release，或者至少是 2.2.x 行 (2.2.6) 中的最新版本。