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 评估上下文的根对象是 消息)不同,在评估时,此入站适配器还未有消息,因为这是它最终使用传输的文件作为其有效负载生成的内容。因此,SpEL 评估上下文的根对象是远程文件的原始名称(一个 字符串)。 入站通道适配器首先将文件检索到本地目录,然后根据轮询配置发送每个文件。从5.0版本开始,您可以在需要新的文件检索时限制从SFTP服务器获取的文件数量。当目标文件较大或在具有持久文件列表筛选器(稍后在本部分中讨论)的群集系统中运行时,这可能很有用。为此,使用`max-fetch-size`。负值(默认值)表示没有限制,并且检索所有匹配的文件。有关更多信息,请参见Inbound Channel Adapters: Controlling Remote File Fetching。从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`策略的实例中(参见Metadata Store)。此过滤器根据文件名和远程修改时间进行匹配。 从 4.0 版本开始,此过滤器需要一个 ConcurrentMetadataStore。当与共享数据存储(例如使用 RedisMetadataStoreRedis)一起使用时,这可以使筛选密钥在多个应用程序或服务器实例之间共享。 从 5.0 版本开始,默认情况下 SftpInboundFileSynchronizer 使用带有内存中 SimpleMetadataStoreSftpPersistentAcceptOnceFileListFilter。此过滤器也与 XML 配置中的 regexpattern 选项一起应用,以及通过 Java DSL 中的 SftpInboundChannelAdapterSpec 应用。你可以使用 CompositeFileListFilter(或 ChainFileListFilter)处理任何其他用例。 上述讨论涉及在检索文件之前对其进行筛选。检索文件后,会将其他过滤器应用到文件系统上的文件。默认情况下,这是一个`AcceptOnceFileListFilter`,如本部分所述,它将状态保留在内存中,并且不考虑文件的修改时间。除非你的应用程序在处理后删除文件,否则该适配器在应用程序重新启动后默认会重新处理磁盘上的文件。 另外,如果你将 filter 配置为使用 SftpPersistentAcceptOnceFileListFilter 并且远程文件时间戳发生更改(导致重新获取),则默认本地过滤器不允许处理此新文件。 有关此过滤器的详细信息以及如何使用它的信息,请参见 Remote Persistent File List Filters。 您可以使用`local-filter`属性来配置本地文件系统筛选器的行为。从4.3.8版本开始,默认情况下配置`FileSystemPersistentAcceptOnceFileListFilter`。此过滤器将已接受的文件名和修改时间戳存储在`MetadataStore`策略的实例中(参见Metadata Store),并检测本地文件修改时间的更改。默认的`MetadataStore`是将状态存储在内存中的`SimpleMetadataStore`。 从 4.1.5 版本开始,这些过滤器都有一个名为 flushOnUpdate 的新属性,它导致它们在每次更新时刷新元数据存储(如果存储实现 Flushable)。

此外,如果您使用分布式 MetadataStore(例如 Redis Metadata Store),您可以拥有相同适配器或应用程序的多个实例,并确保只有一个实例处理一个文件。

实际的本地过滤器是一个包含所提供的过滤器和一个模式过滤器的 CompositeFileListFilter,该模式过滤器防止处理正在下载中的文件(基于 temporary-file-suffix)。文件使用此后缀下载(默认值为 .writing),并且在传输完成后,文件将重命名为其最终名称,使其对过滤器“可见”。 有关这些属性的更多详细信息,请参见 schema。 SFTP 入站通道适配器是轮询使用者。因此,你必须配置轮询器(全局默认值或本地元素)。一旦文件被传输到本地目录,就会生成一个 java.io.File 作为其有效负载类型的消息,并将其发送到由 channel 属性标识的通道。 从 6.2 版本开始,你可以使用 SftpLastModifiedFileListFilter 基于最后修改策略筛选 SFTP 文件。该过滤器可以配置为 age 属性,以便过滤器仅传递比此值更旧的文件。年龄默认为 60 秒,但你应该选择足够大的年龄,以避免过早提取文件(例如,由于网络故障)。请查看其 Javadoc 以了解更多信息。

More on File Filtering and Large Files

有时,监视的(远程)目录中刚刚出现的文件并不完整。通常,此类文件会用一些临时扩展名编写(例如命名为 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"/>

Recovering from Failures

您应该了解适配器的体系结构。文件同步器检索文件,而`FileReadingMessageSource`为每个同步文件发送消息。作为discussed earlier,涉及两个过滤器。`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 版开始,传入通道适配器可以根据生成的文件名在本地建立子目录。那也可以是一个远端的子路径。为了能够根据层次结构支持递归读取本地目录的修改,您现在可以提供带有基于 Files.walk() 算法的新 RecursiveDirectoryScannerFileReadingMessageSource 内部。有关详细信息,请参见 AbstractInboundFileSynchronizingMessageSource.setScanner()。此外,您现在可以使用 setUseWatchService() 选项,将 AbstractInboundFileSynchronizingMessageSource 切换到基于 WatchServiceDirectoryScanner。它还针对所有 WatchEventType 实例进行配置,以对本地目录的任何修改做出反应。先前展示的重新处理示例基于 FileReadingMessageSource.WatchServiceDirectoryScanner 的内置功能,当文件从本地目录中删除(StandardWatchEventKinds.ENTRY_DELETE)时,FileReadingMessageSource.WatchServiceDirectoryScanner 会使用 ResettableFileListFilter.remove()。有关详细信息,请参见 WatchServiceDirectoryScanner

Configuring with Java Configuration

以下 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());
            }

        };
    }

}

Configuring with the 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();
    }
}

Dealing With Incomplete Data

请参阅 @{66}。

提供 SftpSystemMarkerFilePresentFileListFilter 以过滤远程系统上没有对应标记文件的远程文件。有关配置信息,请参见 Javadoc