FTP入站通道适配器
FTP入站通道适配器是一种特殊的监听器,它连接到FTP服务器并监听远程目录事件(例如,创建了新文件),并在此时启动文件传输。以下示例展示了如何配置一个inbound-channel-adapter
<int-ftp:inbound-channel-adapter id="ftpInbound"
channel="ftpChannel"
session-factory="ftpSessionFactory"
auto-create-local-directory="true"
delete-remote-files="true"
filename-pattern="*.txt"
remote-directory="some/remote/path"
remote-file-separator="/"
preserve-timestamp="true"
local-filename-generator-expression="#this.toUpperCase() + '.a'"
scanner="myDirScanner"
local-filter="myFilter"
temporary-file-suffix=".writing"
max-fetch-size="-1"
local-directory=".">
<int:poller fixed-rate="1000"/>
</int-ftp:inbound-channel-adapter>
如上所示配置,您可以使用inbound-channel-adapter
元素配置FTP入站通道适配器,同时提供各种属性的值,例如local-directory
、filename-pattern
(基于简单的模式匹配,而非正则表达式),以及对session-factory
的引用。
默认情况下,传输的文件与原始文件同名。如果您想覆盖此行为,可以设置local-filename-generator-expression
属性,该属性允许您提供一个SpEL表达式来生成本地文件的名称。与出站网关和适配器不同,后者SpEL评估上下文的根对象是一个Message
,此入站适配器在评估时还没有消息,因为最终它会使用传输的文件作为payload生成消息。因此,SpEL评估上下文的根对象是远程文件的原始名称(一个String
)。
入站通道适配器首先检索本地目录的File
对象,然后根据轮询器配置发射每个文件。从版本5.0开始,您现在可以限制在需要检索新文件时从FTP服务器获取的文件数量。当目标文件非常大或在带有持久化文件列表过滤器(稍后讨论)的集群系统中运行时,这可能非常有益。为此目的使用max-fetch-size
。负值(默认值)表示无限制,将检索所有匹配的文件。更多信息请参阅入站通道适配器:控制远程文件获取。自版本5.0以来,您还可以通过设置scanner
属性为inbound-channel-adapter
提供自定义的DirectoryScanner
实现。
从Spring Integration 3.0开始,您可以指定preserve-timestamp
属性(默认为false
)。当设置为true
时,本地文件的修改时间戳会被设置为从服务器获取的值。否则,它会被设置为当前时间。
从版本4.2开始,您可以指定remote-directory-expression
代替remote-directory
,允许您在每次轮询时动态确定目录 — 例如,remote-directory-expression="@myBean.determineRemoteDir()"
。
从版本4.3开始,您可以省略remote-directory
和remote-directory-expression
属性。它们默认为null
。在这种情况下,根据FTP协议,客户端工作目录将用作默认的远程目录。
有时,基于filename-pattern
属性指定的简单模式的文件过滤可能不够。在这种情况下,您可以使用filename-regex
属性来指定正则表达式(例如filename-regex=".*\.test$"
)。此外,如果您需要完全控制,可以使用filter
属性并提供对o.s.i.file.filters.FileListFilter
的任何自定义实现的引用,这是一个用于过滤文件列表的策略接口。此过滤器确定要检索哪些远程文件。您还可以通过使用CompositeFileListFilter
将基于模式的过滤器与其他过滤器(例如AcceptOnceFileListFilter
,以避免同步以前已获取的文件)组合使用。
AcceptOnceFileListFilter
将其状态存储在内存中。如果您希望状态在系统重启后仍然存在,请考虑使用FtpPersistentAcceptOnceFileListFilter
。此过滤器将接受的文件名存储在MetadataStore
策略的实例中(参见元数据存储)。此过滤器根据文件名和远程修改时间进行匹配。
自版本4.0起,此过滤器需要ConcurrentMetadataStore
。当与共享数据存储(例如带有RedisMetadataStore
的Redis)一起使用时,它允许在多个应用程序或服务器实例之间共享过滤器键。
从版本5.0开始,带有内存中SimpleMetadataStore
的FtpPersistentAcceptOnceFileListFilter
默认应用于FtpInboundFileSynchronizer
。在XML配置中,此过滤器也适用于regex
或pattern
选项,以及Java DSL中的FtpInboundChannelAdapterSpec
。任何其他用例可以通过CompositeFileListFilter
(或ChainFileListFilter
)进行管理。
前面的讨论是指在检索文件之前进行过滤。一旦文件被检索到,系统会向文件系统上的文件应用一个额外的过滤器。默认情况下,这是一个AcceptOnceFileListFilter
,如前所述,它在内存中保留状态,并且不考虑文件的修改时间。除非您的应用程序在处理后删除文件,否则适配器在应用程序重启后会默认重新处理磁盘上的文件。
此外,如果您将过滤器配置为使用FtpPersistentAcceptOnceFileListFilter
,并且远程文件的时间戳发生变化(导致它被重新获取),默认的本地过滤器将不允许处理这个新文件。
有关此过滤器及其使用方法的更多信息,请参阅远程持久化文件列表过滤器。
您可以使用local-filter
属性配置本地文件系统过滤器的行为。从版本4.3.8开始,默认配置了FileSystemPersistentAcceptOnceFileListFilter
。此过滤器将接受的文件名和修改时间戳存储在MetadataStore
策略的实例中(参见元数据存储),并检测本地文件修改时间的变化。默认的MetadataStore
是SimpleMetadataStore
,它将状态存储在内存中。
自版本4.1.5以来,这些过滤器有一个新属性(flushOnUpdate
),该属性使得它们在每次更新时刷新元数据存储(如果存储实现了Flushable
)。
此外,如果您使用分布式MetadataStore (例如Redis),您可以拥有同一适配器或应用程序的多个实例,并确保每个文件只处理一次。 |
实际的本地过滤器是一个CompositeFileListFilter
,它包含提供的过滤器和一个模式过滤器,后者用于防止处理正在下载的文件(基于temporary-file-suffix
)。文件会以这个后缀(默认为.writing
)下载,并在传输完成后重命名为最终名称,使其对过滤器“可见”。
remote-file-separator
属性允许您配置文件分隔符,以便在默认的'/'不适用于您的特定环境时使用。
有关这些属性的更多详细信息,请参阅schema。
您还应该了解,FTP入站通道适配器是一个轮询消费者。因此,您必须配置一个轮询器(通过使用全局默认配置或本地子元素)。文件传输完成后,将生成一个以java.io.File
作为payload的消息,并发送到由channel
属性标识的通道。
从版本6.2开始,您可以使用FtpLastModifiedFileListFilter
基于最后修改策略过滤FTP文件。可以为该过滤器配置一个age
属性,以便只有比该值旧的文件才能通过过滤器。默认的age
为60秒,但您应该选择一个足够大的值,以避免过早地获取文件(例如,由于网络故障)。请查阅其Javadoc了解更多信息。
更多关于文件过滤和不完整文件
有时刚出现在受监控的(远程)目录中的文件是不完整的。通常,这样的文件会以临时扩展名(例如somefile.txt.writing
)写入,然后在写入过程完成后重命名。在大多数情况下,您只对完整的文件感兴趣,并希望只过滤完整的文件。为了处理这些场景,您可以使用filename-pattern
、filename-regex
和filter
属性提供的过滤支持。以下示例使用了自定义过滤器实现
<int-ftp:inbound-channel-adapter
channel="ftpChannel"
session-factory="ftpSessionFactory"
filter="customFilter"
local-directory="file:/my_transfers">
remote-directory="some/remote/path"
<int:poller fixed-rate="1000"/>
</int-ftp:inbound-channel-adapter>
<bean id="customFilter" class="org.example.CustomFilter"/>
入站FTP适配器的轮询器配置说明
入站FTP适配器的任务包括两个
-
与远程服务器通信,以便将文件从远程目录传输到本地目录。
-
对于每个传输的文件,生成一个以该文件为payload的消息,并将其发送到由“channel”属性标识的通道。这就是为什么它们被称为“通道适配器”,而不是仅仅是“适配器”。这种适配器的主要工作是生成要发送到消息通道的消息。本质上,第二个任务优先执行,如果您的本地目录中已经有一个或多个文件,它会首先从这些文件生成消息。只有在所有本地文件都被处理完毕后,它才会启动远程通信来检索更多文件。
此外,在配置轮询器上的触发器时,您应该密切注意max-messages-per-poll
属性。对于所有SourcePollingChannelAdapter
实例(包括FTP),其默认值为1
。这意味着,只要处理完一个文件,它就会等待由您的触发器配置确定的下一次执行时间。如果您在local-directory
中有意存放了一个或多个文件,它会在启动与远程FTP服务器的通信之前处理这些文件。此外,如果max-messages-per-poll
设置为1
(默认值),它会按照触发器定义的间隔一次只处理一个文件,本质上工作方式是“一次轮询 === 一个文件”。
对于典型的文件传输用例,您最有可能想要相反的行为:在每次轮询时处理尽可能多的文件,然后才等待下一次轮询。如果是这种情况,请将max-messages-per-poll
设置为-1
。然后,在每次轮询时,适配器会尝试生成尽可能多的消息。换句话说,它会处理本地目录中的所有内容,然后连接到远程目录,将那里所有可供本地处理的文件都传输过来。只有这样,轮询操作才算完成,并且轮询器才会等待下一次执行时间。
或者,您可以将max-messages-per-poll
的值设置为一个正数,表示每次轮询从文件创建的消息数量的上限。例如,值为10
表示在每次轮询时,它尝试处理的文件数量不超过十个。
从故障中恢复
理解适配器的架构非常重要。有一个文件同步器负责获取文件,还有一个FileReadingMessageSource
负责为每个同步的文件发送消息。如前所述,涉及到两个过滤器。filter
属性(和模式)指代远程(FTP)文件列表,以避免获取已经获取过的文件。local-filter
由FileReadingMessageSource
使用,以确定哪些文件将被作为消息发送。
同步器列出远程文件并咨询其过滤器。然后文件被传输。如果在文件传输过程中发生IO错误,任何已经添加到过滤器的文件都会被移除,以便在下一次轮询时可以重新获取。这只适用于过滤器实现了ReversibleFileListFilter
(例如AcceptOnceFileListFilter
)的情况。
如果在同步文件后,下游流程处理文件时发生错误,过滤器不会自动回滚,因此默认情况下失败的文件不会被重新处理。
如果您希望在发生故障后重新处理这些文件,可以使用类似于以下的配置来方便地从过滤器中移除失败的文件
<int-ftp:inbound-channel-adapter id="ftpAdapter"
session-factory="ftpSessionFactory"
channel="requestChannel"
remote-directory-expression="'/ftpSource'"
local-directory="file:myLocalDir"
auto-create-local-directory="true"
filename-pattern="*.txt">
<int:poller fixed-rate="1000">
<int:transactional synchronization-factory="syncFactory" />
</int:poller>
</int-ftp:inbound-channel-adapter>
<bean id="acceptOnceFilter"
class="org.springframework.integration.file.filters.AcceptOnceFileListFilter" />
<int:transaction-synchronization-factory id="syncFactory">
<int:after-rollback expression="payload.delete()" />
</int:transaction-synchronization-factory>
<bean id="transactionManager"
class="org.springframework.integration.transaction.PseudoTransactionManager" />
上述配置适用于任何ResettableFileListFilter
。
从版本5.0开始,入站通道适配器可以在本地构建与生成的本地文件名对应的子目录。这也可以是远程子路径。为了能够根据层次结构支持递归读取本地目录的修改,您现在可以为内部的FileReadingMessageSource
提供一个新的基于Files.walk()
算法的RecursiveDirectoryScanner
。更多信息请参阅AbstractInboundFileSynchronizingMessageSource.setScanner()
。此外,您现在可以使用setUseWatchService()
选项将AbstractInboundFileSynchronizingMessageSource
切换到基于WatchService
的DirectoryScanner
。它也配置了所有WatchEventType
实例,以对本地目录中的任何修改作出反应。前面展示的重新处理示例基于FileReadingMessageSource.WatchServiceDirectoryScanner
的内置功能,即当文件从本地目录中删除(StandardWatchEventKinds.ENTRY_DELETE
)时执行ResettableFileListFilter.remove()
。更多信息请参阅WatchServiceDirectoryScanner
。
使用Java配置
以下Spring Boot应用程序展示了如何使用Java配置配置入站适配器
@SpringBootApplication
public class FtpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(FtpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public SessionFactory<FTPFile> ftpSessionFactory() {
DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
sf.setHost("localhost");
sf.setPort(port);
sf.setUsername("foo");
sf.setPassword("foo");
sf.setTestSession(true);
return new CachingSessionFactory<FTPFile>(sf);
}
@Bean
public FtpInboundFileSynchronizer ftpInboundFileSynchronizer() {
FtpInboundFileSynchronizer fileSynchronizer = new FtpInboundFileSynchronizer(ftpSessionFactory());
fileSynchronizer.setDeleteRemoteFiles(false);
fileSynchronizer.setRemoteDirectory("foo");
fileSynchronizer.setFilter(new FtpSimplePatternFileListFilter("*.xml"));
return fileSynchronizer;
}
@Bean
@InboundChannelAdapter(channel = "ftpChannel", poller = @Poller(fixedDelay = "5000"))
public MessageSource<File> ftpMessageSource() {
FtpInboundFileSynchronizingMessageSource source =
new FtpInboundFileSynchronizingMessageSource(ftpInboundFileSynchronizer());
source.setLocalDirectory(new File("ftp-inbound"));
source.setAutoCreateLocalDirectory(true);
source.setLocalFilter(new AcceptOnceFileListFilter<File>());
source.setMaxFetchSize(1);
return source;
}
@Bean
@ServiceActivator(inputChannel = "ftpChannel")
public MessageHandler handler() {
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
System.out.println(message.getPayload());
}
};
}
}
使用Java DSL配置
以下Spring Boot应用程序展示了如何使用Java DSL配置入站适配器
@SpringBootApplication
public class FtpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(FtpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public IntegrationFlow ftpInboundFlow() {
return IntegrationFlow
.from(Ftp.inboundAdapter(this.ftpSessionFactory)
.preserveTimestamp(true)
.remoteDirectory("foo")
.regexFilter(".*\\.txt$")
.localFilename(f -> f.toUpperCase() + ".a")
.localDirectory(new File("d:\\ftp_files")),
e -> e.id("ftpInboundAdapter")
.autoStartup(true)
.poller(Pollers.fixedDelay(5000)))
.handle(m -> System.out.println(m.getPayload()))
.get();
}
}