文件分割器
FileSplitter
在 4.1.2 版本中添加,其命名空间支持在 4.2 版本中添加。FileSplitter
根据 BufferedReader.readLine()
将文本文件分割成单独的行。默认情况下,分割器使用 Iterator
在从文件中读取行时逐行发出。将 iterator
属性设置为 false
会使其在发出消息之前将所有行读入内存。这样做的一种用例是,您可能想在发送任何包含行的消息之前检测文件上的 I/O 错误。然而,这只适用于相对较短的文件。
入站载荷可以是 File
、String
(文件路径)、InputStream
或 Reader
。其他载荷类型将原样发出。
以下列表显示了配置 FileSplitter
的可能方式
-
Java DSL
-
Kotlin DSL
-
Java
-
XML
@SpringBootApplication
public class FileSplitterApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(FileSplitterApplication.class)
.web(false)
.run(args);
}
@Bean
public IntegrationFlow fileSplitterFlow() {
return IntegrationFlow
.from(Files.inboundAdapter(tmpDir.getRoot())
.filter(new ChainFileListFilter<File>()
.addFilter(new AcceptOnceFileListFilter<>())
.addFilter(new ExpressionFileListFilter<>(
new FunctionExpression<File>(f -> "foo.tmp".equals(f.getName()))))))
.split(Files.splitter()
.markers()
.charset(StandardCharsets.US_ASCII)
.firstLineAsHeader("fileHeader")
.applySequence(true))
.channel(c -> c.queue("fileSplittingResultChannel"))
.get();
}
}
@Bean
fun fileSplitterFlow() =
integrationFlow(
Files.inboundAdapter(tmpDir.getRoot())
.filter(
ChainFileListFilter<File?>()
.addFilter(AcceptOnceFileListFilter())
.addFilter(ExpressionFileListFilter(FunctionExpression { f: File? -> "foo.tmp" == f!!.name }))
)
) {
split(
Files.splitter()
.markers()
.charset(StandardCharsets.US_ASCII)
.firstLineAsHeader("fileHeader")
.applySequence(true)
)
channel { queue("fileSplittingResultChannel") }
}
@Splitter(inputChannel="toSplitter")
@Bean
public MessageHandler fileSplitter() {
FileSplitter splitter = new FileSplitter(true, true);
splitter.setApplySequence(true);
splitter.setOutputChannel(outputChannel);
return splitter;
}
<int-file:splitter id="splitter" (1)
iterator="" (2)
markers="" (3)
markers-json="" (4)
apply-sequence="" (5)
requires-reply="" (6)
charset="" (7)
first-line-as-header="" (8)
input-channel="" (9)
output-channel="" (10)
send-timeout="" (11)
auto-startup="" (12)
order="" (13)
phase="" /> (14)
1 | 分割器的 bean 名称。 |
2 | 设置为 true (默认值)以使用迭代器,或者设置为 false 以在发送行之前将文件加载到内存中。 |
3 | 设置为 true 以在文件数据之前和之后发出文件开始和文件结束标记消息。标记是载荷为 FileSplitter.FileMarker 的消息(其 mark 属性包含 START 和 END 值)。当您在下游流中按顺序处理文件,并且某些行被过滤掉时,可以使用标记。它们使下游处理能够知道文件何时已完全处理完毕。此外,一个包含 START 或 END 的 file_marker 头会被添加到这些消息中。END 标记包含行计数。如果文件为空,则只发出 START 和 END 标记,其 lineCount 为 0 。默认值为 false 。当设置为 true 时,apply-sequence 默认为 false 。另请参阅 markers-json (下一个属性)。 |
4 | 当 markers 为 true 时,将其设置为 true 可将 FileMarker 对象转换为 JSON 字符串。(底层使用 SimpleJsonSerializer )。 |
5 | 设置为 false 可禁用在消息中包含 sequenceSize 和 sequenceNumber 头。默认值为 true ,除非 markers 为 true 。当为 true 且 markers 为 true 时,标记会包含在序列中。当为 true 且 iterator 为 true 时,sequenceSize 头会被设置为 0 ,因为大小未知。 |
6 | 设置为 true 可在文件没有行时抛出 RequiresReplyException 。默认值为 false 。 |
7 | 设置读取文本数据到 String 载荷时使用的字符集名称。默认值是平台字符集。 |
8 | 第一行作为头信息包含在为剩余行发出的消息中的头名称。自 5.0 版本起。 |
9 | 设置用于向分割器发送消息的输入通道。 |
10 | 设置消息发送到的输出通道。 |
11 | 设置发送超时。仅当 output-channel 可能阻塞时适用,例如满的 QueueChannel 。 |
12 | 设置为 false 可禁用在上下文刷新时自动启动分割器。默认值为 true 。 |
13 | 如果 input-channel 是 <publish-subscribe-channel/> ,则设置此端点的顺序。 |
14 | 设置分割器的启动阶段(当 auto-startup 为 true 时使用)。 |
FileSplitter
还可以将任何基于文本的 InputStream
分割成行。从 4.3 版本开始,当与 FTP 或 SFTP 流式入站通道适配器,或使用 stream
选项检索文件的 FTP 或 SFTP 出站网关结合使用时,当文件完全被消费时,分割器会自动关闭支持该流的会话。有关这些功能的更多信息,请参见 FTP 流式入站通道适配器 和 SFTP 流式入站通道适配器,以及 FTP 出站网关 和 SFTP 出站网关。
使用 Java 配置时,还有一个额外的构造函数可用,如下例所示
public FileSplitter(boolean iterator, boolean markers, boolean markersJson)
当 markersJson
为 true 时,标记表示为 JSON 字符串(使用 SimpleJsonSerializer
)。
5.0 版本引入了 firstLineAsHeader
选项,用于指定内容的第一行是头(例如 CSV 文件中的列名)。传递给此属性的参数是头名称,在该名称下,第一行作为头信息包含在为剩余行发出的消息中。此行不包含在序列头中(如果 applySequence
为 true),也不包含在与 FileMarker.END
关联的 lineCount
中。注意:从 5.5 版本开始,lineCount
也作为 FileHeaders.LINE_COUNT
包含在 FileMarker.END
消息的头中,因为 FileMarker
可以被序列化为 JSON。如果文件只包含头行,则该文件被视为空文件,因此在分割期间只发出 FileMarker
实例(如果启用了标记 — 否则,不发出任何消息)。默认情况下(如果未设置头名称),第一行被视为数据,并成为第一个发出的消息的载荷。
如果您需要更复杂的逻辑来从文件内容中提取头(不是第一行,不是行的全部内容,不是某个特定的头等等),可以考虑在 FileSplitter
之前使用 头增强器。请注意,已移至头的行可能会在下游被过滤掉,而不参与正常的内容处理。
分割文件后的幂等下游处理
当 apply-sequence
为 true 时,分割器会在 SEQUENCE_NUMBER
头中添加行号(当 markers
为 true 时,标记被计为行)。行号可以与 幂等接收器 一起使用,以避免在重新启动后重新处理行。
例如
@Bean
public ConcurrentMetadataStore store() {
return new ZookeeperMetadataStore();
}
@Bean
public MetadataStoreSelector selector() {
return new MetadataStoreSelector(
message -> message.getHeaders().get(FileHeaders.ORIGINAL_FILE, File.class)
.getAbsolutePath(),
message -> message.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)
.toString(),
store())
.compareValues(
(oldVal, newVal) -> Integer.parseInt(oldVal) < Integer.parseInt(newVal));
}
@Bean
public IdempotentReceiverInterceptor idempotentReceiverInterceptor() {
return new IdempotentReceiverInterceptor(selector());
}
@Bean
public IntegrationFlow flow() {
...
.split(new FileSplitter())
...
.handle("lineHandler", e -> e.advice(idempotentReceiverInterceptor()))
...
}