java - 使用 Splitter 和其他一些组件后如何保持相同的相关 ID？

Question

我正在使用 Spring Integration XPath Splitters 和 Header Enrichers，如下所示:

    <!-- split the folders-->
    <int-xml:xpath-splitter input-channel="PQAdditionalContactHistory-Split-Filtered-Folders-Channel" output-channel="PQAdditionalContactHistory-Split-Filtered-Folders-Channel" >
    <int-xml:xpath-expression expression="//response/results/PQCallHistory/row"/>
    </int-xml:xpath-splitter>

<int:chain  input-channel="PQAdditionalContactHistory-Split-Filtered-Folders-Channel" output-channel="PQContactHistory-InputChannel" >
    <int:service-activator  ref="msgHandler" method="buildMessageFromExtSysResponse" />
    <int-xml:xslt-transformer xsl-resource="${stylesheet.PQAdditionalContactHistory-To-PQContactHistory-Request}" />
</int:chain>   

<int:channel id="PQContactHistory-InputChannel" />              
    <int:chain input-channel="PQContactHistory-InputChannel" output-channel="PQContactHistory-OutputChannel">

        <int-xml:xpath-splitter>
            <int-xml:xpath-expression expression="//PQCallHistory"  namespace-map="xmlMessageNamespace" />
        </int-xml:xpath-splitter>

        <int-xml:xslt-transformer xsl-resource="${stylesheet.PQContactHistoryStoredProcData}"  />

        <!-- Store the original payload in header for future purpose -->
        <int:header-enricher default-overwrite="true"  should-skip-nulls="true"  >
            <int:header name="${headerNames.originalPayload}" expression="payload" />   
        </int:header-enricher>  
    </int:chain>            

这里，计划是使用聚合器，并在使用这些拆分消息执行一些操作后合并所有拆分消息。令人惊讶的是，所有拆分消息列表并未立即到达聚合器。当我进一步调查以了解根本原因时，我发现拆分消息的相关 ID 在第二个拆分器之后发生了更改。但是，我需要由第一个拆分器拆分的所有消息列表作为一个 block 到达聚合器端。关于如何实现这一目标有什么想法吗？如果我的问题不清楚，请告诉我。

我使用默认<aggregator>行为:<int:aggregator input-channel="test" ref="xmlAggregator" method="aggreagateAdditionalFolders"></int:aggregator> 。

Answer 1

这是事实，因为任何 splitter提供SequenceDetails ( sequenceId 、 sequenceSize 等)默认在其范围内并将它们存储在 MessageHeaders 中通过stack方式。

不幸的是<int-xml:xpath-splitter> doesn't support apply-sequence="false"还没有。

但是您可以按照该 Pull 请求的 JIRA 解决方法进行操作。

这只是克服您的用例的一方面。

另一件事是围绕<aggregator> 。有需要知道你如何correlate和release在那里分组。

我可能猜你只是依赖默认的 <aggregator>行为 - 处理这个问题SequenceDetails header 。

这不太好。第一个之后xpath-splitter ，sequenceSize header 已获取作为多个回复消息的值。既然你要去split他们每人一次，没有信息有多少元素会到达 <aggregator> 。

关于sequenceNumber也是如此。 header ，表示 <aggregator> 的标记是否将消息应用到组。

所以，请显示您的聚合器配置。之后我们将尝试共同提供解决方案。

无论如何，在这种情况下，您应该提供一些自定义的 ReleaseStrategy对于聚合器。CorrelationStrategy可能会保留为默认值。

更新

示例如何配置XPathMessageSplitter作为通用 bean:

<int:service-activator>
    <beans:bean class="org.springframework.integration.xml.splitter.XPathMessageSplitter">
        <beans:constructor-arg type="org.springframework.xml.xpath.XPathExpression">
            <int-xml:xpath-expression expression="//PQCallHistory" namespace-map="xmlMessageNamespace" />
        </beans:constructor-arg>
    </beans:bean>
</int:service-activator>