Spring Integration - 聚合分散的消息答案

【问题标题】：Spring Integration - Aggregate Scattered MessagesSpring Integration - 聚合分散的消息
【发布时间】：2012-12-19 01:43:43
【问题描述】：

我有一个 Spring Integration 应用程序，它将传入的文件放到一个通道上。从那里我希望能够将相同的文件发送到两个不同的处理管道（一个归档到 S3，另一个解析内容），然后有一个下游组件可以识别两者何时都已成功处理，从而删除实际的本地文件。

语义就像我需要一个拆分器/聚合器，但我需要复制它而不是拆分消息。

有什么方法可以使用可用的组件来实现这一点，还是需要一些自定义类？

【问题讨论】：

标签： java spring file spring-integration

【解决方案1】：

是的，<publish-subscribe-channel/>（带有apply-sequence="true"）的工作方式类似于拆分器 - 但是频道的两个订阅者都将获得相同的File 对象。默认情况下，这两个分支将串行执行，但如果您想并行处理，您可以引入一个 ExecutorChannel。

如果您希望每个订阅者获得不同的File 对象，您可以添加一个转换器...

<transformer ... expression="new java.io.File(payload.absolutePath)" />

【讨论】：

加里，你一如既往地是明星！
在此之后，我认为您在使用聚合器来测试两个过程何时完成方面是正确的。聚合您的 pub-sub 频道侦听器在完成后将发出的一些唯一键（文件名？）并设置“#this.size() eq 2”的发布策略表达式。确保配置 MessageGroupStoreReaper 以处理其中一个进程失败的情况。
其实apply-sequence会添加sequence size、sequence number和标准correlationId headers。
表示可以使用标准的关联和发布策略。