读取文件
FileReadingMessageSource 可用于从文件系统消费文件。它是 MessageSource 的一个实现，用于从文件系统目录创建消息。以下示例演示如何配置 FileReadingMessageSource
<bean id="pollableFileSource"
class="org.springframework.integration.file.inbound.FileReadingMessageSource"
p:directory="${input.directory}"/>为了防止为某些文件创建消息，您可以提供一个 FileListFilter。默认情况下，我们使用以下过滤器
-
IgnoreHiddenFileListFilter -
AcceptOnceFileListFilter
IgnoreHiddenFileListFilter 确保隐藏文件不会被处理。请注意，隐藏文件的确切定义是系统相关的。例如，在基于 UNIX 的系统中，以句点字符开头的文件被认为是隐藏文件。另一方面，Microsoft Windows 有一个专用的文件属性来指示隐藏文件。
|
版本 4.2 引入了 |
AcceptOnceFileListFilter 确保文件只从目录中拾取一次。
|
自版本 4.0 以来，此过滤器需要一个 自版本 4.1.5 以来，此过滤器有一个新属性（ |
持久文件列表过滤器现在有一个布尔属性 forRecursion。将此属性设置为 true，还会设置 alwaysAcceptDirectories，这意味着出站网关上的递归操作（ls 和 mget）现在每次都会遍历完整的目录树。这是为了解决目录树深处的变化未被检测到的问题。此外，forRecursion=true 会导致文件的完整路径用作元数据存储键；这解决了如果具有相同名称的文件在不同目录中多次出现时过滤器无法正常工作的问题。重要提示：这意味着在顶层目录下的文件中，持久元数据存储中的现有键将找不到。因此，此属性默认为 false；这可能会在未来的版本中更改。
以下示例配置了一个带过滤器的 FileReadingMessageSource
<bean id="pollableFileSource"
class="org.springframework.integration.file.inbound.FileReadingMessageSource"
p:inputDirectory="${input.directory}"
p:filter-ref="customFilterBean"/>读取文件的一个常见问题是，文件可能在准备好之前就被检测到（也就是说，其他进程可能仍在写入文件）。默认的 AcceptOnceFileListFilter 并不能阻止这种情况。在大多数情况下，如果文件写入进程在文件准备好读取后立即重命名每个文件，则可以防止这种情况。一个只接受已准备好文件（可能基于已知后缀）的 filename-pattern 或 filename-regex 过滤器，与默认的 AcceptOnceFileListFilter 组合使用，可以处理这种情况。CompositeFileListFilter 启用这种组合，如下例所示
<bean id="pollableFileSource"
class="org.springframework.integration.file.inbound.FileReadingMessageSource"
p:inputDirectory="${input.directory}"
p:filter-ref="compositeFilter"/>
<bean id="compositeFilter"
class="org.springframework.integration.file.filters.CompositeFileListFilter">
<constructor-arg>
<list>
<bean class="o.s.i.file.filters.AcceptOnceFileListFilter"/>
<bean class="o.s.i.file.filters.RegexPatternFileListFilter">
<constructor-arg value="^test.*$"/>
</bean>
</list>
</constructor-arg>
</bean>如果无法使用临时名称创建文件并重命名为最终名称，Spring Integration 提供另一种替代方案。版本 4.2 添加了 LastModifiedFileListFilter。此过滤器可以配置一个 age 属性，以便只有早于此值的文件才能通过过滤器。年龄默认为 60 秒，但您应该选择一个足够大的年龄，以避免过早地拾取文件（例如，由于网络故障）。以下示例演示如何配置 LastModifiedFileListFilter
<bean id="filter" class="org.springframework.integration.file.filters.LastModifiedFileListFilter">
<property name="age" value="120" />
</bean>从版本 4.3.7 开始，引入了一个 ChainFileListFilter（CompositeFileListFilter 的扩展），以允许后续过滤器只查看上一个过滤器结果的场景。（使用 CompositeFileListFilter，所有过滤器都会看到所有文件，但它只通过所有过滤器都通过的文件）。新行为所需的示例是 LastModifiedFileListFilter 和 AcceptOnceFileListFilter 的组合，当我们不希望在经过一段时间后才接受文件时。使用 CompositeFileListFilter，由于 AcceptOnceFileListFilter 在第一次传递时看到所有文件，因此当其他过滤器传递时，它不会在以后传递。当模式过滤器与查找辅助文件以指示文件传输完成的自定义过滤器组合时，CompositeFileListFilter 方法很有用。模式过滤器可能只传递主文件（例如 something.txt），但“完成”过滤器需要查看（例如）something.done 是否存在。
假设我们有文件 a.txt、a.done 和 b.txt。
模式过滤器只通过 a.txt 和 b.txt，而“完成”过滤器看到所有三个文件，只通过 a.txt。组合过滤器的最终结果是只释放 a.txt。
使用 ChainFileListFilter，如果链中的任何过滤器返回空列表，则不调用其余过滤器。 |
版本 5.0 引入了 ExpressionFileListFilter，用于针对文件作为上下文评估根对象执行 SpEL 表达式。为此，所有用于文件处理（本地和远程）的 XML 组件，以及现有的 filter 属性，都提供了 filter-expression 选项，如下例所示
<int-file:inbound-channel-adapter
directory="${inputdir}"
filter-expression="name matches '.text'"
auto-startup="false"/>版本 5.0.5 引入了 DiscardAwareFileListFilter 实现，它们对被拒绝的文件感兴趣。为此，此类过滤器实现应通过 addDiscardCallback(Consumer<File>) 提供回调。在框架中，此功能用于 FileReadingMessageSource.WatchServiceDirectoryScanner，结合 LastModifiedFileListFilter。与常规的 DirectoryScanner 不同，WatchService 根据目标文件系统上的事件提供文件进行处理。在轮询这些文件的内部队列时，LastModifiedFileListFilter 可能会因为它们相对于其配置的 age 太年轻而丢弃它们。因此，我们丢失了文件以供将来可能的考虑。丢弃回调钩子允许我们将文件保留在内部队列中，以便在后续轮询中可以对其进行 age 检查。CompositeFileListFilter 还实现了一个 DiscardAwareFileListFilter，并将其丢弃回调填充到所有 DiscardAwareFileListFilter 委托中。
由于 CompositeFileListFilter 将文件与所有委托进行匹配，因此 discardCallback 可能会为同一个文件调用多次。 |
从版本 5.1 开始，FileReadingMessageSource 不再检查目录是否存在，也不再创建它，直到调用其 start()（通常通过包装 SourcePollingChannelAdapter）。以前，当引用目录时（例如从测试中，或当权限稍后应用时），没有简单的方法可以防止操作系统权限错误。
与 LastModifiedFileListFilter 相反，从版本 6.5 开始引入了 RecentFileListFilter 策略。它是 AbstractRecentFileListFilter 的本地文件系统扩展。默认情况下，它接受不早于 1 天的文件。请参阅其其他实现以获取相应的远程文件协议。
消息头
从版本 5.0 开始，FileReadingMessageSource（除了作为轮询 File 的 payload）还会将以下标头填充到出站 Message
-
FileHeaders.FILENAME: 要发送的文件的File.getName()。可用于后续的重命名或复制逻辑。 -
FileHeaders.ORIGINAL_FILE:File对象本身。通常，此标头由框架组件（如 拆分器 或 转换器）在丢失原始File对象时自动填充。但是，为了与任何其他自定义用例保持一致和方便，此标头对于访问原始文件可能很有用。 -
FileHeaders.RELATIVE_PATH: 引入了一个新标头，用于表示文件路径中相对于扫描根目录的部分。当需要将源目录层次结构恢复到其他位置时，此标头可能很有用。为此，可以配置DefaultFileNameGenerator（参见 "`生成文件名）以使用此标头。
目录扫描和轮询
FileReadingMessageSource 不会立即为目录中的文件生成消息。它使用内部队列存储由 scanner 返回的“符合条件的文件”。scanEachPoll 选项用于确保内部队列在每次轮询时都使用最新的输入目录内容进行刷新。默认情况下（scanEachPoll = false），FileReadingMessageSource 在再次扫描目录之前清空其队列。这种默认行为在减少对目录中大量文件的扫描时特别有用。但是，在需要自定义排序的情况下，考虑将此标志设置为 true 的影响很重要。文件处理的顺序可能不像预期的那样。默认情况下，队列中的文件按其自然（path）顺序处理。即使队列中已有文件，通过扫描添加的新文件也会插入到适当的位置以保持自然顺序。要自定义顺序，FileReadingMessageSource 可以接受 Comparator<File> 作为构造函数参数。它由内部（PriorityBlockingQueue）使用，根据业务需求重新排序其内容。因此，要按特定顺序处理文件，您应该向 FileReadingMessageSource 提供一个比较器，而不是对自定义 DirectoryScanner 生成的列表进行排序。
版本 5.0 引入了 RecursiveDirectoryScanner 以执行文件树遍历。该实现基于 Files.walk(Path start, int maxDepth, FileVisitOption… options) 功能。根目录（DirectoryScanner.listFiles(File)）参数从结果中排除。所有其他子目录的包含和排除都基于目标 FileListFilter 实现。例如，SimplePatternFileListFilter 默认情况下会过滤掉目录。有关详细信息，请参见 AbstractDirectoryAwareFileListFilter 及其实现。
从版本 5.5 开始，Java DSL 的 FileInboundChannelAdapterSpec 有一个方便的 recursive(boolean) 选项，可以在目标 FileReadingMessageSource 中使用 RecursiveDirectoryScanner 而不是默认的扫描器。 |
从版本 7.0 开始，FileReadingMessageSource 可以使用 SpEL 表达式配置其 directory 属性。每次请求新扫描时都会评估此表达式。这种逻辑在输入目录在扫描完成后应更改或基于时间戳轮换的场景中可能很有用。
命名空间支持
通过使用文件特定的命名空间，可以简化文件读取的配置。为此，请使用以下模板
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-file="http://www.springframework.org/schema/integration/file"
xsi:schemaLocation="http://www.springframework.org/schema/beans
https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/file
https://www.springframework.org/schema/integration/file/spring-integration-file.xsd">
</beans>在此命名空间中，您可以将 FileReadingMessageSource 简化并将其封装在入站通道适配器中，如下所示
<int-file:inbound-channel-adapter id="filesIn1"
directory="file:${input.directory}" prevent-duplicates="true" ignore-hidden="true"/>
<int-file:inbound-channel-adapter id="filesIn2"
directory="file:${input.directory}"
filter="customFilterBean" />
<int-file:inbound-channel-adapter id="filesIn3"
directory="file:${input.directory}"
filename-pattern="test*" />
<int-file:inbound-channel-adapter id="filesIn4"
directory="file:${input.directory}"
filename-regex="test[0-9]+\.txt" />第一个通道适配器示例依赖于默认的 FileListFilter 实现
-
IgnoreHiddenFileListFilter（不处理隐藏文件） -
AcceptOnceFileListFilter（防止重复）
因此，您也可以省略 prevent-duplicates 和 ignore-hidden 属性，因为它们默认为 true。
|
Spring Integration 4.2 引入了 |
第二个通道适配器示例使用自定义过滤器，第三个使用 filename-pattern 属性添加基于 AntPathMatcher 的过滤器，第四个使用 filename-regex 属性向 FileReadingMessageSource 添加基于正则表达式模式的过滤器。filename-pattern 和 filename-regex 属性与常规的 filter 引用属性互斥。但是，您可以使用 filter 属性引用 CompositeFileListFilter 实例，该实例结合了任意数量的过滤器，包括一个或多个基于模式的过滤器，以满足您的特定需求。
当多个进程从同一个目录读取时，您可能需要锁定文件以防止它们并发地被拾取。为此，您可以使用 FileLocker。有一个基于 java.nio 的实现可用，但也可以实现自己的锁定方案。nio 锁定器可以按如下方式注入
<int-file:inbound-channel-adapter id="filesIn"
directory="file:${input.directory}" prevent-duplicates="true">
<int-file:nio-locker/>
</int-file:inbound-channel-adapter>您可以按如下方式配置自定义锁定器
<int-file:inbound-channel-adapter id="filesIn"
directory="file:${input.directory}" prevent-duplicates="true">
<int-file:locker ref="customLocker"/>
</int-file:inbound-channel-adapter>当文件入站适配器配置了锁定器时，它负责在文件被允许接收之前获取锁定。它不承担解锁文件的责任。如果您已经处理了文件并让锁定挂起，则会出现内存泄漏。如果这是一个问题，您应该在适当的时候自己调用 FileLocker.unlock(File file)。 |
当过滤和锁定文件不足以满足需求时，您可能需要完全控制文件列表的方式。要实现此类需求，您可以使用 DirectoryScanner 的实现。此扫描器允许您精确确定每次轮询中列出的文件。这也是 Spring Integration 内部用于将 FileListFilter 实例和 FileLocker 连接到 FileReadingMessageSource 的接口。您可以将自定义 DirectoryScanner 注入到 <int-file:inbound-channel-adapter/> 的 scanner 属性上，如下例所示
<int-file:inbound-channel-adapter id="filesIn" directory="file:${input.directory}"
scanner="customDirectoryScanner"/>这样做让您可以完全自由地选择排序、列表和锁定策略。
同样重要的是要理解，过滤器（包括 patterns、regex、prevent-duplicates 等）和 locker 实例实际上是由 scanner 使用的。适配器上设置的任何这些属性随后都会注入到内部 scanner 中。对于外部 scanner 的情况，FileReadingMessageSource 上禁止使用所有过滤器和锁定器属性。它们必须在自定义 DirectoryScanner 上指定（如果需要）。换句话说，如果您将 scanner 注入到 FileReadingMessageSource 中，您应该在该 scanner 上提供 filter 和 locker，而不是在 FileReadingMessageSource 上。
默认情况下，DefaultDirectoryScanner 使用 IgnoreHiddenFileListFilter 和 AcceptOnceFileListFilter。要阻止它们的使用，您可以配置自己的过滤器（例如 AcceptAllFileListFilter），甚至将其设置为 null。 |
WatchServiceDirectoryScanner
FileReadingMessageSource.WatchServiceDirectoryScanner 在新文件添加到目录时依赖文件系统事件。在初始化期间，目录被注册以生成事件。初始文件列表也在初始化期间构建。在遍历目录树时，遇到的任何子目录也会被注册以生成事件。在第一次轮询中，返回遍历目录的初始文件列表。在后续轮询中，返回来自新创建事件的文件。如果添加了新的子目录，其创建事件将用于遍历新的子树以查找现有文件并注册任何新发现的子目录。
当程序的内部事件 queue 没有像目录修改事件发生那样快速地被排出时，WatchKey 会出现一个问题。如果队列大小超过，会发出 StandardWatchEventKinds.OVERFLOW 以指示可能丢失了一些文件系统事件。在这种情况下，根目录会被完全重新扫描。为了避免重复，请考虑使用适当的 FileListFilter（例如 AcceptOnceFileListFilter）或在处理完成后删除文件。 |
可以通过 FileReadingMessageSource.use-watch-service 选项启用 WatchServiceDirectoryScanner，该选项与 scanner 选项互斥。将为提供的 directory 填充一个内部 FileReadingMessageSource.WatchServiceDirectoryScanner 实例。
此外，现在 WatchService 轮询逻辑可以跟踪 StandardWatchEventKinds.ENTRY_MODIFY 和 StandardWatchEventKinds.ENTRY_DELETE。
如果您需要跟踪现有文件以及新文件的修改，则应在 FileListFilter 中实现 ENTRY_MODIFY 事件逻辑。否则，来自这些事件的文件将以相同的方式处理。
ResettableFileListFilter 实现会捕获 ENTRY_DELETE 事件。因此，它们的文件将用于 remove() 操作。当启用此事件时，诸如 AcceptOnceFileListFilter 等过滤器将删除文件。因此，如果出现同名文件，它会通过过滤器并作为消息发送。
为此，引入了 watch-events 属性（FileReadingMessageSource.setWatchEvents(WatchEventType… watchEvents)）。(WatchEventType 是 FileReadingMessageSource 中的一个公共内部枚举。) 通过此选项，我们可以对新文件使用一个下游流逻辑，对修改过的文件使用另一个逻辑。以下示例演示如何配置不同的逻辑以在同一目录中创建和修改事件
值得一提的是，ENTRY_DELETE 事件与被监视目录的子目录的重命名操作有关。更具体地说，与旧目录名相关的 ENTRY_DELETE 事件先于通知新（重命名）目录的 ENTRY_CREATE 事件。在某些操作系统（如 Windows）上，必须注册 ENTRY_DELETE 事件才能处理这种情况。否则，在文件资源管理器中重命名被监视的子目录可能会导致在该子目录中检测不到新文件。
<int-file:inbound-channel-adapter id="newFiles"
directory="${input.directory}"
use-watch-service="true"/>
<int-file:inbound-channel-adapter id="modifiedFiles"
directory="${input.directory}"
use-watch-service="true"
filter="acceptAllFilter"
watch-events="MODIFY"/> <!-- The default is CREATE. -->从版本 6.1 开始，FileReadingMessageSource 暴露了两个新的 WatchService 相关选项
-
watchMaxDepth-Files.walkFileTree(Path root, Set attributes, int maxDepth, FileVisitor visitor)API 的参数； -
watchDirPredicate- 一个Predicate<Path>，用于测试扫描树中的目录是否应被遍历并注册到WatchService和配置的监视事件类型。
限制内存消耗
您可以使用 HeadDirectoryScanner 限制内存中保留的文件数量。这在扫描大型目录时非常有用。通过 XML 配置，可以通过在入站通道适配器上设置 queue-size 属性来启用此功能。
在版本 4.2 之前，此设置与任何其他过滤器的使用不兼容。任何其他过滤器（包括 prevent-duplicates="true"）都会覆盖用于限制大小的过滤器。
|
使用 通常，在这种情况下，您应该删除已处理的文件，而不是使用 |
使用 Java 配置
以下 Spring Boot 应用程序展示了如何使用 Java 配置配置出站适配器的示例
@SpringBootApplication
public class FileReadingJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(FileReadingJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public MessageChannel fileInputChannel() {
return new DirectChannel();
}
@Bean
@InboundChannelAdapter(value = "fileInputChannel", poller = @Poller(fixedDelay = "1000"))
public MessageSource<File> fileReadingMessageSource() {
FileReadingMessageSource source = new FileReadingMessageSource();
source.setDirectory(new File(INBOUND_PATH));
source.setFilter(new SimplePatternFileListFilter("*.txt"));
return source;
}
@Bean
@Transformer(inputChannel = "fileInputChannel", outputChannel = "processFileChannel")
public FileToStringTransformer fileToStringTransformer() {
return new FileToStringTransformer();
}
}使用 Java DSL 进行配置
以下 Spring Boot 应用程序演示了如何使用 Java DSL 配置出站适配器
@SpringBootApplication
public class FileReadingJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(FileReadingJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public IntegrationFlow fileReadingFlow() {
return IntegrationFlow
.from(Files.inboundAdapter(new File(INBOUND_PATH))
.patternFilter("*.txt"),
e -> e.poller(Pollers.fixedDelay(1000)))
.transform(Files.toStringTransformer())
.channel("processFileChannel")
.get();
}
}“跟踪”文件
另一个流行的用例是从文件的末尾（或尾部）获取“行”，并在添加新行时捕获它们。提供了两种实现。第一种是 OSDelegatingFileTailingMessageProducer，它使用本机 tail 命令（在具有该命令的操作系统上）。这通常是这些平台上最有效的实现。对于没有 tail 命令的操作系统，第二种实现 ApacheCommonsFileTailingMessageProducer 使用 Apache commons-io 的 Tailer 类。
在这两种情况下，文件系统事件，例如文件不可用和其他事件，都通过正常的 Spring 事件发布机制发布为 ApplicationEvent 实例。此类事件的示例包括以下内容
[message=tail: cannot open '/tmp/somefile' for reading:
No such file or directory, file=/tmp/somefile]
[message=tail: '/tmp/somefile' has become accessible, file=/tmp/somefile]
[message=tail: '/tmp/somefile' has become inaccessible:
No such file or directory, file=/tmp/somefile]
[message=tail: '/tmp/somefile' has appeared;
following end of new file, file=/tmp/somefile]例如，在文件轮换时，可能会发生前面示例中所示的事件序列。
从版本 5.0 开始，当文件在 idleEventInterval 期间没有数据时，会发出 FileTailingIdleEvent。以下示例展示了此类事件的样子
[message=Idle timeout, file=/tmp/somefile] [idle time=5438]并非所有支持 tail 命令的平台都提供这些状态消息。 |
从这些端点发出的消息具有以下标头
-
FileHeaders.ORIGINAL_FILE：File对象 -
FileHeaders.FILENAME：文件名（File.getName()）
在 5.0 版本之前，FileHeaders.FILENAME 标头包含文件绝对路径的字符串表示。您现在可以通过调用原始文件标头上的 getAbsolutePath() 来获取该字符串表示。 |
以下示例创建了一个具有默认选项（'-F -n 0'，表示从当前末尾跟踪文件名）的本机适配器。
<int-file:tail-inbound-channel-adapter id="native"
channel="input"
task-executor="exec"
file="/tmp/foo"/>以下示例创建了一个带有 '-F -n +0' 选项（表示跟踪文件名，发出所有现有行）的本机适配器。
<int-file:tail-inbound-channel-adapter id="native"
channel="input"
native-options="-F -n +0"
task-executor="exec"
file-delay=10000
file="/tmp/foo"/>如果 tail 命令失败（在某些平台上，文件缺失会导致 tail 失败，即使指定了 -F），该命令将每 10 秒重试一次。
默认情况下，本机适配器从标准输出捕获内容并将其作为消息发送。它们还从标准错误捕获内容以引发事件。从版本 4.3.6 开始，您可以通过将 enable-status-reader 设置为 false 来丢弃标准错误事件，如下例所示
<int-file:tail-inbound-channel-adapter id="native"
channel="input"
enable-status-reader="false"
task-executor="exec"
file="/tmp/foo"/>在以下示例中，IdleEventInterval 设置为 5000，这意味着，如果五秒钟内没有写入行，则每五秒触发一次 FileTailingIdleEvent
<int-file:tail-inbound-channel-adapter id="native"
channel="input"
idle-event-interval="5000"
task-executor="exec"
file="/tmp/somefile"/>这在需要停止适配器时很有用。
以下示例创建了一个 Apache commons-io Tailer 适配器，该适配器每两秒检查一次文件中的新行，并每十秒检查一次缺失文件的存在
<int-file:tail-inbound-channel-adapter id="apache"
channel="input"
task-executor="exec"
file="/tmp/bar"
delay="2000"
end="false" (1)
reopen="true" (2)
file-delay="10000"/>| 1 | 文件从开头（end="false"）而不是结尾（这是默认设置）进行跟踪。 |
| 2 | 文件为每个块重新打开（默认是保持文件打开）。 |
指定 delay、end 或 reopen 属性会强制使用 Apache commons-io 适配器，并使 native-options 属性不可用。 |
