对于最新的稳定版本，请使用 Spring Integration 6.5.1！spring-doc.cadn.net.cn

文件聚合器

从 5.5 版开始，一个FileAggregator被引入以涵盖FileSplitter启用 START/END 标记时的用例。为方便起见，该FileAggregator实现所有三种序列详细信息策略：spring-doc.cadn.net.cn

这HeaderAttributeCorrelationStrategy使用FileHeaders.FILENAME属性用于关联键计算。当在FileSplitter，它不会填充序列详细信息标头，因为 START/END 标记消息也包含在序列大小中。这FileHeaders.FILENAME仍会为发出的每一行填充，包括 START/END 标记消息。spring-doc.cadn.net.cn
这FileMarkerReleaseStrategy- 检查FileSplitter.FileMarker.Mark.END消息，然后比较FileHeaders.LINE_COUNT标头值，组大小减去2 - FileSplitter.FileMarker实例。它还实现了方便的GroupConditionProvider联系方式conditionSupplier函数要用于AbstractCorrelatingMessageHandler. 有关详细信息，请参阅消息组条件。spring-doc.cadn.net.cn
这FileAggregatingMessageGroupProcessor只需删除FileSplitter.FileMarker消息，并将其余消息收集到列表有效负载中进行生成。spring-doc.cadn.net.cn

以下列表显示了配置FileAggregator:spring-doc.cadn.net.cn

@Bean
public IntegrationFlow fileSplitterAggregatorFlow(TaskExecutor taskExecutor) {
    return f -> f
            .split(Files.splitter()
                    .markers()
                    .firstLineAsHeader("firstLine"))
            .channel(c -> c.executor(taskExecutor))
            .filter(payload -> !(payload instanceof FileSplitter.FileMarker),
                    e -> e.discardChannel("aggregatorChannel"))
            .<String, String>transform(String::toUpperCase)
            .channel("aggregatorChannel")
            .aggregate(new FileAggregator())
            .channel(c -> c.queue("resultChannel"));
}

@Bean
fun fileSplitterAggregatorFlow(taskExecutor: TaskExecutor?) =
    integrationFlow {
        split(Files.splitter().markers().firstLineAsHeader("firstLine"))
        channel { executor(taskExecutor) }
        filter<Any>({ it !is FileMarker }) { discardChannel("aggregatorChannel") }
        transform(String::toUpperCase)
        channel("aggregatorChannel")
        aggregate(FileAggregator())
        channel { queue("resultChannel") }
    }

@serviceActivator(inputChannel="toAggregateFile")
@Bean
public AggregatorFactoryBean fileAggregator() {
    AggregatorFactoryBean aggregator = new AggregatorFactoryBean();
    aggregator.setProcessorBean(new FileAggregator());
    aggregator.setOutputChannel(outputChannel);
    return aggregator;
}

<int:chain input-channel="input" output-channel="output">
    <int-file:splitter markers="true"/>
    <int:aggregator>
        <bean class="org.springframework.integration.file.aggregator.FileAggregator"/>
    </int:aggregator>
</int:chain>

如果默认行为的FileAggregator不满足目标逻辑，建议使用单个策略配置聚合器端点。看FileAggregatorJavaDocs 了解更多信息。spring-doc.cadn.net.cn