文件聚合器
从 5.5 版本开始,引入了 FileAggregator
,用于在启用 START/END 标记时覆盖 FileSplitter
的另一种用例。为方便起见,FileAggregator
实现了所有三种序列详细信息策略
-
使用带有
FileHeaders.FILENAME
属性的HeaderAttributeCorrelationStrategy
来计算关联键。当FileSplitter
上启用标记时,它不会填充序列详细信息头,因为 START/END 标记消息也包含在序列大小中。FileHeaders.FILENAME
仍然会为发出的每一行填充,包括 START/END 标记消息。 -
FileMarkerReleaseStrategy
- 检查组中是否存在FileSplitter.FileMarker.Mark.END
消息,然后将FileHeaders.LINE_COUNT
头值与组大小减去2
(即FileSplitter.FileMarker
实例数)进行比较。它还实现了一个方便的GroupConditionProvider
契约,用于在AbstractCorrelatingMessageHandler
中使用conditionSupplier
函数。更多信息请参阅 消息组条件。 -
FileAggregatingMessageGroupProcessor
仅从组中移除FileSplitter.FileMarker
消息,并将其余消息收集到列表 payload 中进行生成。
以下列表显示了配置 FileAggregator
的可能方法
-
Java DSL
-
Kotlin DSL
-
Java
-
XML
@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
的默认行为不满足目标逻辑,建议使用独立的策略配置聚合器端点。更多信息请参阅 FileAggregator
的 JavaDocs。