Spring Integration - 按组并行文件处理

Question

我正在尝试通过一个简单的任务对 Spring Integration 进行实验。我有一个文件夹，我可以在其中获取传入文件。这些文件以组 ID 命名。 我希望按顺序处理同一 groupId 中的所有文件，但可以并行处理具有不同 groupId 的文件。

我开始整理这样的配置：

<int:service-activator input-channel="filesInChannel"
    output-channel="outputChannelAdapter">
    <bean class="com.ingestion.FileProcessor" />
</int:service-activator>

<int:channel id="filesInChannel" />

<int-file:inbound-channel-adapter id="inputChannelAdapter"
    channel="filesInChannel" directory="${in.file.path}" prevent-duplicates="true"
    filename-pattern="${file.pattern}">

    <int:poller id="poller" fixed-rate="1" task-executor="executor"/>
</int-file:inbound-channel-adapter>

<int-file:outbound-channel-adapter id="outputChannelAdapter" directory="${ok.file.path}" delete-source-files="true"/>

<task:executor id="executor" pool-size="10"/>

这是用 10 个线程处理所有传入的文件。我需要哪些步骤来按 groupId 拆分文件并让它们为每个 groupId 处理一个线程？

谢谢。

Answer 1

假设组 id 数量有限，您可以为每个组使用不同的适配器（使用单个线程；全部馈入同一通道）；每个都有不同的模式。

或者您可以创建一个自定义 FileListFilter 并使用某种线程关联性将每个组中的文件分配给特定线程，过滤器仅返回该线程的文件。