跳到主要内容
版本:7.0.2

SFTP 入站通道适配器

DeepSeek V3 中英对照 SFTP Inbound Channel Adapter

SFTP入站通道适配器是一种特殊的监听器,它会连接到服务器并监听远程目录事件(例如新文件创建),并在事件发生时启动文件传输。以下示例展示了如何配置SFTP入站通道适配器:

<int-sftp:inbound-channel-adapter id="sftpAdapterAutoCreate"
session-factory="sftpSessionFactory"
channel="requestChannel"
filename-pattern="*.txt"
remote-directory="/foo/bar"
preserve-timestamp="true"
local-directory="file:target/foo"
auto-create-local-directory="true"
local-filename-generator-expression="#this.toUpperCase() + '.a'"
scanner="myDirScanner"
local-filter="myFilter"
temporary-file-suffix=".writing"
max-fetch-size="-1"
delete-remote-files="false">
<int:poller fixed-rate="1000"/>
</int-sftp:inbound-channel-adapter>

前面的配置示例展示了如何为各种属性提供值,包括以下内容:

  • local-directory:文件将被传输到的本地目录位置

  • remote-directory:文件将被传输的远程源目录

  • session-factory:对我们先前配置的 bean 的引用

默认情况下,传输的文件会保留与原始文件相同的名称。若需覆盖此行为,可设置 local-filename-generator-expression 属性,该属性允许通过 SpEL 表达式生成本地文件名。与出站网关和适配器不同(其 SpEL 评估上下文的根对象为 Message),此入站适配器在评估时尚未生成消息——因为消息最终是以传输的文件作为其负载生成的。因此,SpEL 评估上下文的根对象是远程文件的原始名称(即 String 类型)。

入站通道适配器首先将文件检索到本地目录,然后根据轮询器配置逐个发送文件。从版本 5.0 开始,当需要检索新文件时,您可以限制从 SFTP 服务器获取的文件数量。当目标文件较大或在具有持久文件列表过滤器的集群系统中运行时,此功能非常有用,本节稍后将对此进行讨论。为此,请使用 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()"

有时,仅通过 filename-pattern 属性指定的简单模式进行文件过滤可能不够用。如果遇到这种情况,你可以使用 filename-regex 属性来指定一个正则表达式,例如 filename-regex=".*\.test$"。如果你需要完全控制,可以使用 filter 属性来引用一个自定义的 org.springframework.integration.file.filters.FileListFilter 实现,这是一个用于过滤文件列表的策略接口。该过滤器决定了哪些远程文件会被获取。你还可以通过使用 CompositeFileListFilter,将基于模式的过滤器与其他过滤器(例如 AcceptOnceFileListFilter,以避免同步之前已获取的文件)组合使用。

AcceptOnceFileListFilter 将其状态存储在内存中。如果您希望状态在系统重启后得以保留,请考虑改用 SftpPersistentAcceptOnceFileListFilter。此过滤器将已接受的文件名存储在 MetadataStore 策略的实例中(参见 元数据存储)。此过滤器根据文件名和远程修改时间进行匹配。

自 4.0 版本起,此过滤器需要一个 ConcurrentMetadataStore。当与共享数据存储(例如使用 RedisMetadataStoreRedis)结合使用时,该配置允许过滤器密钥在多个应用程序或服务器实例之间共享。

从版本 5.0 开始,SftpInboundFileSynchronizer 默认应用了带有内存 SimpleMetadataStoreSftpPersistentAcceptOnceFileListFilter。该过滤器也会与 XML 配置中的 regexpattern 选项一起应用,同时通过 Java DSL 中的 SftpInboundChannelAdapterSpec 应用。您可以通过使用 CompositeFileListFilter(或 ChainFileListFilter)来处理任何其他用例。

上述讨论涉及在检索文件之前进行过滤。一旦文件被检索到,还会对文件系统上的文件应用额外的过滤器。默认情况下,这是一个 AcceptOnceFileListFilter,正如本节所讨论的,它在内存中保留状态,并且不考虑文件的修改时间。除非您的应用程序在处理后删除文件,否则适配器在应用程序重启后默认会重新处理磁盘上的文件。

此外,如果将 filter 配置为使用 SftpPersistentAcceptOnceFileListFilter,并且远程文件的时间戳发生更改(导致文件被重新获取),默认的本地过滤器将不允许处理这个新文件。

有关此过滤器及其使用方式的更多信息,请参阅远程持久文件列表过滤器

您可以使用 local-filter 属性来配置本地文件系统过滤器的行为。从 4.3.8 版本开始,默认配置了一个 FileSystemPersistentAcceptOnceFileListFilter。该过滤器将已接受的文件名和修改时间戳存储在 MetadataStore 策略的实例中(参见 Metadata Store),并检测本地文件修改时间的变化。默认的 MetadataStoreSimpleMetadataStore,它在内存中存储状态。

自 4.1.5 版本起,这些过滤器新增了一个名为 flushOnUpdate 的属性,该属性会使它们在每次更新时刷新元数据存储(前提是该存储实现了 Flushable 接口)。

important

此外,如果你使用分布式 MetadataStore(例如 Redis 元数据存储),你可以拥有同一适配器或应用程序的多个实例,并确保有且仅有一个实例处理一个文件。

实际使用的本地过滤器是一个 ChainFileListFilter,它包含一个模式过滤器,用于防止处理正在下载的文件(基于 temporary-file-suffix)以及提供的过滤器。文件下载时会带有此后缀(默认为 .writing),传输完成后文件会被重命名为最终名称,从而使其对过滤器“可见”。

有关这些属性的更多详细信息,请参阅 schema

SFTP入站通道适配器是一种轮询消费者。因此,您必须配置轮询器(可以是全局默认值,也可以是本地元素)。文件传输到本地目录后,会生成一个以 java.io.File 作为其负载类型的消息,并发送到由 channel 属性标识的通道。

从 6.2 版本开始,你可以使用 SftpLastModifiedFileListFilter 基于最后修改时间策略来筛选 SFTP 文件。该过滤器可以配置一个 age 属性,以便只有修改时间早于此值的文件才能通过过滤器。age 的默认值为 60 秒,但你应该选择一个足够大的值,以避免过早地选取文件(例如,由于网络故障)。更多信息请查阅其 Javadoc。

相比之下,从版本6.5开始,引入了SftpRecentFileListFilter,它只接受那些不早于指定age的文件。

更多关于文件过滤和大文件的内容

有时,监控(远程)目录中刚出现的文件可能尚未完整。通常,这类文件会以临时扩展名(例如名为 something.txt.writing 的文件使用 .writing 扩展名)写入,然后在写入过程完成后重命名。大多数情况下,开发者只对已完成的文件感兴趣,并希望仅筛选这些文件。为处理此类场景,您可以使用 filename-patternfilename-regexfilter 属性提供的筛选支持。如果需要自定义筛选器实现,可以通过设置 filter 属性在适配器中包含引用。以下示例展示了具体操作方法:

<int-sftp:inbound-channel-adapter id="sftpInbondAdapter"
channel="receiveChannel"
session-factory="sftpSessionFactory"
filter="customFilter"
local-directory="file:/local-test-dir"
remote-directory="/remote-test-dir">
<int:poller fixed-rate="1000" max-messages-per-poll="10" task-executor="executor"/>
</int-sftp:inbound-channel-adapter>

<bean id="customFilter" class="org.foo.CustomFilter"/>

从故障中恢复

你需要理解适配器的架构。文件同步器负责获取文件,而 FileReadingMessageSource 会为每个同步的文件发出消息。如前文所述,其中涉及两个过滤器。filter 属性(及其模式)用于远程(SFTP)文件列表,以避免获取已获取过的文件。FileReadingMessageSource 则使用 local-filter 来确定哪些文件应作为消息发送。

同步器会列出远程文件并参考其过滤器。随后,文件将被传输。如果在文件传输过程中发生IO错误,任何已添加到过滤器中的文件都会被移除,以便它们有资格在下一次轮询时重新获取。这仅适用于实现了 ReversibleFileListFilter(例如 AcceptOnceFileListFilter)的过滤器。

如果在同步文件后,下游流程处理文件时发生错误,过滤器不会自动回滚,因此默认情况下不会重新处理失败的文件。

若希望在失败后重新处理此类文件,您可以使用类似以下配置,以便从过滤器中移除失败文件:

<int-sftp:inbound-channel-adapter id="sftpAdapter"
session-factory="sftpSessionFactory"
channel="requestChannel"
remote-directory-expression="'/sftpSource'"
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-sftp: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 提供一个新的 RecursiveDirectoryScanner,该扫描器基于 Files.walk() 算法。更多信息请参阅 AbstractInboundFileSynchronizingMessageSource.setScanner()。此外,您现在可以通过使用 setUseWatchService() 选项将 AbstractInboundFileSynchronizingMessageSource 切换到基于 WatchServiceDirectoryScanner。该扫描器还配置为对所有 WatchEventType 实例作出反应,以响应本地目录中的任何修改。前面展示的重处理示例基于 FileReadingMessageSource.WatchServiceDirectoryScanner 的内置功能,当文件从本地目录中删除(StandardWatchEventKinds.ENTRY_DELETE)时,该扫描器会使用 ResettableFileListFilter.remove()。更多信息请参阅 WatchServiceDirectoryScanner

使用 Java 配置进行配置

以下Spring Boot应用程序展示了如何使用Java配置入站适配器的示例:

@SpringBootApplication
public class SftpJavaApplication {

public static void main(String[] args) {
new SpringApplicationBuilder(SftpJavaApplication.class)
.web(false)
.run(args);
}

@Bean
public SessionFactory<SftpClient.DirEntry> sftpSessionFactory() {
DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
factory.setHost("localhost");
factory.setPort(port);
factory.setUser("foo");
factory.setPassword("foo");
factory.setAllowUnknownKeys(true);
factory.setTestSession(true);
return new CachingSessionFactory<>(factory);
}

@Bean
public SftpInboundFileSynchronizer sftpInboundFileSynchronizer() {
SftpInboundFileSynchronizer fileSynchronizer = new SftpInboundFileSynchronizer(sftpSessionFactory());
fileSynchronizer.setDeleteRemoteFiles(false);
fileSynchronizer.setRemoteDirectory("foo");
fileSynchronizer.setFilter(new SftpSimplePatternFileListFilter("*.xml"));
return fileSynchronizer;
}

@Bean
@InboundChannelAdapter(channel = "sftpChannel", poller = @Poller(fixedDelay = "5000"))
public MessageSource<File> sftpMessageSource() {
SftpInboundFileSynchronizingMessageSource source =
new SftpInboundFileSynchronizingMessageSource(sftpInboundFileSynchronizer());
source.setLocalDirectory(new File("sftp-inbound"));
source.setAutoCreateLocalDirectory(true);
source.setLocalFilter(new AcceptOnceFileListFilter<File>());
source.setMaxFetchSize(1);
return source;
}

@Bean
@ServiceActivator(inputChannel = "sftpChannel")
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 SftpJavaApplication {

public static void main(String[] args) {
new SpringApplicationBuilder(SftpJavaApplication.class)
.web(false)
.run(args);
}

@Bean
public IntegrationFlow sftpInboundFlow() {
return IntegrationFlow
.from(Sftp.inboundAdapter(this.sftpSessionFactory)
.preserveTimestamp(true)
.remoteDirectory("foo")
.regexFilter(".*\\.txt$")
.localFilenameExpression("#this.toUpperCase() + '.a'")
.localDirectory(new File("sftp-inbound")),
e -> e.id("sftpInboundAdapter")
.autoStartup(true)
.poller(Pollers.fixedDelay(5000)))
.handle(m -> System.out.println(m.getPayload()))
.get();
}
}

处理不完整数据

SftpSystemMarkerFilePresentFileListFilter 用于过滤远程系统中没有对应标记文件的远程文件。有关配置信息,请参阅 Javadoc