SFTP 适配器

Spring Integration 支持通过 SFTP 进行文件传输操作。

安全文件传输协议 (SFTP) 是一种网络协议,可让您通过任何可靠的流在 Internet 上的两台计算机之间传输文件。

SFTP 协议需要一个安全通道,例如 SSH,以及在整个 SFTP 会话期间对客户端身份的可见性。

Spring Integration 通过提供三个客户端端点支持通过 SFTP 发送和接收文件:入站通道适配器、出站通道适配器和出站网关。它还提供了方便的命名空间配置来定义这些客户端组件。

您需要将此依赖项包含到您的项目中:

Maven
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-sftp</artifactId>
    <version>5.5.13</version>
</dependency>
Gradle
compile "org.springframework.integration:spring-integration-sftp:5.5.13"

要在您的 xml 配置中包含 SFTP 命名空间,请在根元素中包含以下属性:

xmlns:int-sftp="http://www.springframework.org/schema/integration/sftp"
xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp
    https://www.springframework.org/schema/integration/sftp/spring-integration-sftp.xsd"

SFTP 会话工厂

从 3.0 版开始,默认情况下不再缓存会话。请参阅SFTP 会话缓存

在配置 SFTP 适配器之前,您必须配置 SFTP 会话工厂。您可以使用常规 bean 定义配置 SFTP 会话工厂,如以下示例所示:

<beans:bean id="sftpSessionFactory"
    class="org.springframework.integration.sftp.session.DefaultSftpSessionFactory">
    <beans:property name="host" value="localhost"/>
    <beans:property name="privateKey" value="classpath:META-INF/keys/sftpTest"/>
    <beans:property name="privateKeyPassphrase" value="springIntegration"/>
    <beans:property name="port" value="22"/>
    <beans:property name="user" value="kermit"/>
</beans:bean>

每次适配器从其请求会话对象时SessionFactory,都会创建一个新的 SFTP 会话。在幕后,SFTP 会话工厂依赖JSch库来提供 SFTP 功能。

但是,Spring Integration 也支持 SFTP 会话的缓存。有关详细信息,请参阅SFTP 会话缓存

JSch 支持通过连接到服务器的多个通道(操作)。默认情况下,Spring Integration 会话工厂为每个通道使用单独的物理连接。从 Spring Integration 3.0 开始,您可以配置会话工厂(使用布尔构造函数 arg - default false)以使用与服务器的单个连接并JSch在该单个连接上创建多个通道。

使用此功能时,您必须将会话工厂包装在缓存会话工厂中,如下所述,以便在操作完成时连接不会物理关闭。

如果缓存被重置,只有在最后一个通道关闭时会话才会断开。

新操作获取会话时,如果发现断开连接,则刷新连接。

如果您遇到连接问题并希望跟踪会话创建并查看轮询了哪些会话,您可以通过将记录器设置为TRACE级别来启用跟踪(例如,log4j.category.org.springframework.integration.sftp=TRACE)。请参阅SFTP/JSCH 日志记录

现在您需要做的就是将此 SFTP 会话工厂注入您的适配器。

为 SFTP 会话工厂提供值的更实用的方法是使用 Spring 的属性占位符支持

配置属性

以下列表描述了DefaultSftpSessionFactory.

isSharedSession(构造函数参数)::When true,使用单个连接,并且JSch Channels是多路复用的。它默认为false.

clientVersion::让您设置客户端版本属性。它的默认值取决于底层的 JSch 版本,但它看起来像:SSH-2.0-JSCH-0.1.45

enableDaemonThread::If true,所有线程都是守护线程。如果设置为false,则使用普通的非守护线程。此属性在基础会话上设置。在那里,此属性默认为false.

host::您要连接的主机的 URL。必需的。

hostKeyAlias::设置主机密钥别名,用于将主机密钥与已知主机列表进行比较。

knownHostsResource::指定用于主机密钥存储库的文件资源。该文件与 OpenSSH 的文件格式相同,是必需的,如果为 false known_hosts,则必须预先填充。allowUnknownKeys

password:: 对远程主机进行身份验证的密码。如果未提供密码,则privateKey需要该属性。如果设置是不允许的userInfo。密码是从该对象获得的。

port:: 建立 SFTP 连接的端口。如果未指定,则此值默认为22。如果指定,此属性必须是正数。

privateKey::让您设置一个资源,该资源表示用于对远程主机进行身份验证的私钥的位置。如果privateKey未提供,则password需要该属性。

privateKeyPassphrase::私钥的密码。如果设置userInfoprivateKeyPassphrase是不允许的。密码是从该对象获得的。可选的。

proxy::允许指定基于 JSch 的代理。如果设置,代理对象用于通过代理创建到远程主机的连接。有关配置代理的便捷方法,请参阅代理工厂 Bean

serverAliveCountMax::指定服务器活动消息的数量,这些消息在断开连接之前发送而没有来自服务器的任何回复。如果未设置,则此属性默认为1

serverAliveInterval::设置发送服务器活动消息之前的超时间隔(以毫秒为单位),以防没有从服务器接收到消息。

sessionConfig::通过使用Properties,您可以在底层JSch Session上设置额外的配置设置。

socketFactory::让你传入一个SocketFactory. 套接字工厂用于创建到目标主机的套接字。当使用代理时,套接字工厂被传递给代理。默认情况下,使用纯 TCP 套接字。

timeout::timeout 属性用作套接字超时参数,以及默认的连接超时时间。默认为0,这意味着不会发生超时。

user::要使用的远程用户。必需的。

allowUnknownKeys::设置为true允许连接到具有未知(或更改)密钥的主机。它的默认值为“假”。只有在没有userInfo提供时才应用它。如果false,则需要预先填充的knownHosts文件。

userInfo::设置UserInfo在身份验证期间使用的自定义。特别是,promptYesNo()当接收到未知(或更改)的主机密钥时调用。另请参阅allowUnknownKeys。提供 aUserInfo时,会从中获取password和 私钥passphrase,不能设置离散 passwordprivateKeyPassphrase属性。

代理工厂 Bean

Jsch提供一种通过 HTTP 或 SOCKS 代理连接到服务器的机制。要使用此功能,请配置Proxy并提供对 的引用DefaultSftpSessionFactory,如前所述。三个实现由Jsch: HTTPSOCKS4和提供SOCKS5。Spring Integration 4.3FactoryBean通过允许属性注入引入了这些代理的轻松配置,如以下示例所示:

<bean id="proxySocks5" class="org.springframework.integration.sftp.session.JschProxyFactoryBean">
    <constructor-arg value="SOCKS5" />
    <constructor-arg value="${sftp.proxy.address}" />
    <constructor-arg value="${sftp.proxy.port}" />
    <constructor-arg value="${sftp.proxy.user}" />
    <constructor-arg value="${sftp.proxy.pw}" />
</bean>

<bean id="sessionFactory"
          class="org.springframework.integration.sftp.session.DefaultSftpSessionFactory" >
    ...
    <property name="proxy" ref="proxySocks5" />
    ...
</bean>

委托会话工厂

4.2 版引入了DelegatingSessionFactory,它允许在运行时选择实际的会话工厂。在调用 SFTP 端点之前,您可以调用setThreadKey()工厂以将密钥与当前线程相关联。然后使用该密钥查找要使用的实际会话工厂。您可以在使用后通过调用来清除密钥clearThreadKey()

我们添加了便利方法,以便您可以更轻松地从消息流中执行此操作,如以下示例所示:

<bean id="dsf" class="org.springframework.integration.file.remote.session.DelegatingSessionFactory">
    <constructor-arg>
        <bean class="o.s.i.file.remote.session.DefaultSessionFactoryLocator">
            <!-- delegate factories here -->
        </bean>
    </constructor-arg>
</bean>

<int:service-activator input-channel="in" output-channel="c1"
        expression="@dsf.setThreadKey(#root, headers['factoryToUse'])" />

<int-sftp:outbound-gateway request-channel="c1" reply-channel="c2" ... />

<int:service-activator input-channel="c2" output-channel="out"
        expression="@dsf.clearThreadKey(#root)" />
使用会话缓存时(请参阅SFTP 会话缓存),应缓存每个委托。您不能缓存DelegatingSessionFactory自身。

从 5.0.7 版本开始,DelegatingSessionFactory可以与 a 结合使用RotatingServerAdvice来轮询多个服务器;请参阅入站通道适配器:轮询多个服务器和目录

SFTP 会话缓存

从 Spring Integration 3.0 版开始,默认情况下不再缓存会话。cache-sessions端点不再支持该属性。如果您希望缓存会话,则必须使用 a CachingSessionFactory(参见下一个示例)。

在 3.0 之前的版本中,默认情况下会自动缓存会话。有一个cache-sessions属性可用于禁用自动缓存,但该解决方案没有提供配置其他会话缓存属性的方法。例如,您不能限制创建的会话数。为了支持该要求和其他配置选项,我们添加了一个CachingSessionFactory. 它提供sessionCacheSizesessionWaitTimeout属性。顾名思义,该sessionCacheSize属性控制工厂在其缓存中维护的活动会话数(默认为无界)。如果sessionCacheSize已达到阈值,则任何获取另一个会话的尝试都会阻塞,直到其中一个缓存会话变得可用或直到会话的等待时间到期(默认等待时间为Integer.MAX_VALUE)。这sessionWaitTimeout属性启用等待时间的配置。

如果您希望缓存会话,请配置默认会话工厂(如前所述),然后将其包装CachingSessionFactory在您可以提供这些附加属性的实例中。以下示例显示了如何执行此操作:

<bean id="sftpSessionFactory"
    class="org.springframework.integration.sftp.session.DefaultSftpSessionFactory">
    <property name="host" value="localhost"/>
</bean>

<bean id="cachingSessionFactory"
    class="org.springframework.integration.file.remote.session.CachingSessionFactory">
    <constructor-arg ref="sftpSessionFactory"/>
    <constructor-arg value="10"/>
    <property name="sessionWaitTimeout" value="1000"/>
</bean>

前面的示例创建了一个,CachingSessionFactorysessionCacheSize设置为一秒(1000 毫秒)。10sessionWaitTimeout

从 Spring Integration 3.0 版开始,CachingConnectionFactory提供了一个resetCache()方法。调用时,所有空闲会话都立即关闭,使用中的会话在返回缓存时关闭。使用isSharedSession=true时,通道关闭,共享会话仅在最后一个通道关闭时关闭。新的会话请求会根据需要建立新的会话。

从 5.1 版开始,CachingSessionFactory有一个新属性testSession。如果为 true,会话将通过执行stat(getHome())命令进行测试,以确保它仍然处于活动状态;如果没有,它将从缓存中删除;如果缓存中没有活动会话,则会创建一个新会话。

使用RemoteFileTemplate

Spring Integration 3.0 版为对象提供了一个新的抽象SftpSession。该模板提供了发送、检索(作为InputStream)、删除和重命名文件的方法。此外,我们提供了一种execute方法让调用者在会话上运行多个操作。在所有情况下,模板都会负责可靠地关闭会话。有关更多信息,请参阅SFTP 的子类的JavadocRemoteFileTemplateSftpRemoteFileTemplate:。

我们在 4.1 版中添加了其他方法,包括getClientInstance(). 它提供对底层的访问ChannelSftp,从而可以访问低级 API。

5.0 版引入了该RemoteFileOperations.invoke(OperationsCallback<F, T> action)方法。此方法允许RemoteFileOperations在同一个 thread-bounded 范围内调用多个调用Session。当您需要将多个高级操作RemoteFileTemplate作为一个工作单元执行时,这很有用。例如,AbstractRemoteFileOutboundGateway将它与mput命令实现一起使用,我们put对提供的目录中的每个文件执行操作,并递归地对其子目录执行操作。有关更多信息,请参阅Javadoc

SFTP 入站通道适配器

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 评估上下文的根对象是 a 的出站网关和适配器不同Message,此入站适配器在评估时还没有消息,因为这是它最终使用传输的文件作为其有效负载生成的。因此,SpEL 评估上下文的根对象是远程文件的原始名称 (a String)。

入站通道适配器首先将文件检索到本地目录,然后根据轮询器配置发出每个文件。从 5.0 版开始,您可以在需要检索新文件时限制从 SFTP 服务器获取的文件数量。当目标文件很大或在具有持久文件列表过滤器的集群系统中运行时,这可能很有用,这将在本节后面讨论。用于max-fetch-size此目的。负值(默认值)表示没有限制并检索所有匹配的文件。有关详细信息,请参阅入站通道适配器:控制远程文件获取。从 5.0 版开始,您还可以通过设置属性DirectoryScanner来提供自定义实现。inbound-channel-adapterscanner

从 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,这是一个用于过滤文件列表的策略接口。此过滤器确定检索哪些远程文件。您还可以AcceptOnceFileListFilter通过使用CompositeFileListFilter.

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

从 4.0 版开始,此过滤器需要ConcurrentMetadataStore. 当与共享数据存储(例如)Redis一起使用时RedisMetadataStore,这允许过滤器键在多个应用程序或服务器实例之间共享。

从 5.0 版开始,默认情况下将SftpPersistentAcceptOnceFileListFilter带有内存的. 此过滤器也与 XML 配置中的or选项以及Java DSL 中的选项一起应用。您可以使用(或)处理任何其他用例。SimpleMetadataStoreSftpInboundFileSynchronizerregexpatternSftpInboundChannelAdapterSpecCompositeFileListFilterChainFileListFilter

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

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

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

您可以使用该local-filter属性来配置本地文件系统过滤器的行为。从版本 4.3.8 开始,FileSystemPersistentAcceptOnceFileListFilter默认配置 a。此过滤器将接受的文件名和修改的时间戳存储在MetadataStore策略实例中(请参阅元数据存储),并检测本地文件修改时间的更改。默认值MetadataStoreSimpleMetadataStore在内存中存储状态的 a。

从 4.1.5 版本开始,这些过滤器有一个名为 的新属性flushOnUpdate,这会导致它们在每次更新时刷新元数据存储(如果存储实现了Flushable)。

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

实际的本地过滤器是一个CompositeFileListFilter包含提供的过滤器和一个模式过滤器,用于阻止处理正在下载的文件(基于temporary-file-suffix)。使用此后缀(默认为.writing)下载文件,并在传输完成后将文件重命名为其最终名称,使它们对过滤器“可见”。

有关这些属性的更多详细信息,请参见架构

SFTP 入站通道适配器是轮询消费者。因此,您必须配置轮询器(全局默认值或本地元素)。一旦文件被传输到本地目录,java.io.File就会生成一条带有其有效负载类型的消息,并将其发送到该channel属性标识的通道。

更多关于文件过滤和大文件

有时,刚刚出现在受监视(远程)目录中的文件是不完整的。通常,此类文件使用一些临时扩展名写入(例如.writing在名为 的文件上something.txt.writing),然后在写入过程完成后重命名。在大多数情况下,开发人员只对完整的文件感兴趣,并且只想过滤那些文件。要处理这些情况,您可以使用 、 和 属性提供的filename-pattern过滤filename-regex支持filter。如果需要自定义过滤器实现,可以通过设置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) 文件列表,以避免获取已经获取的文件。使用来确定要作为消息发送的文件FileReadingMessageSourcelocal-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带有新的内部目录。有关更多信息,请参阅。此外,您现在可以使用选项切换到-based 。它还为所有实例配置,以对本地目录中的任何修改做出反应。前面显示的重新处理示例基于 的内置功能,当从本地目录中删除文件 ( ) 时使用该功能。看RecursiveDirectoryScannerFiles.walk()AbstractInboundFileSynchronizingMessageSource.setScanner()AbstractInboundFileSynchronizingMessageSourceWatchServiceDirectoryScannersetUseWatchService()WatchEventTypeFileReadingMessageSource.WatchServiceDirectoryScannerResettableFileListFilter.remove()StandardWatchEventKinds.ENTRY_DELETEWatchServiceDirectoryScanner了解更多信息。

使用 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<LsEntry> 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<LsEntry>(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 IntegrationFlows
            .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

SFTP 流式入站通道适配器

4.3 版引入了流式入站通道适配器。此适配器生成带有类型有效负载的消息InputStream,让您无需写入本地文件系统即可获取文件。由于会话保持打开状态,因此消费应用程序负责在文件被消费后关闭会话。会话在closeableResource标题 ( IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE) 中提供。标准框架组件,例如FileSplitterand StreamTransformer,会自动关闭会话。有关这些组件的更多信息,请参阅文件拆分器流转换器。以下示例显示如何配置 SFTP 流式入站通道适配器:

<int-sftp:inbound-streaming-channel-adapter id="ftpInbound"
            channel="ftpChannel"
            session-factory="sessionFactory"
            filename-pattern="*.txt"
            filename-regex=".*\.txt"
            filter="filter"
            filter-expression="@myFilterBean.check(#root)"
            remote-file-separator="/"
            comparator="comparator"
            max-fetch-size="1"
            remote-directory-expression="'foo/bar'">
        <int:poller fixed-rate="1000" />
</int-sftp:inbound-streaming-channel-adapter>

您只能使用filename-patternfilename-regexfilter或中的一个filter-expression

从 5.0 版开始,默认情况下,适配器通过使用基于 in-memorySftpStreamingMessageSource来防止远程文件的重复。默认情况下,此过滤器也与文件名模式(或正则表达式)一起应用。如果需要允许重复,可以使用. 您可以使用(或)处理任何其他用例。稍后显示的 Java 配置显示了一种在处理后删除远程文件以避免重复的技术。 SftpPersistentAcceptOnceFileListFilterSimpleMetadataStoreAcceptAllFileListFilterCompositeFileListFilterChainFileListFilter

有关SftpPersistentAcceptOnceFileListFilter以及如何使用它的更多信息,请参阅远程持久文件列表过滤器

当需要获取时,您可以使用该max-fetch-size属性来限制每次轮询获取的文件数。在集群环境中运行时将其设置为1并使用持久过滤器。有关详细信息,请参阅入站通道适配器:控制远程文件获取

适配器将远程目录和文件名放在标题中(FileHeaders.REMOTE_DIRECTORYFileHeaders.REMOTE_FILE,分别)。从 5.0 版开始,FileHeaders.REMOTE_FILE_INFO标头提供了额外的远程文件信息(以 JSON 格式)。如果在to上设置fileInfoJson属性,则标头包含一个对象。可以通过该方法访问底层Jsch库提供的对象。当您使用 XML 配置时,该属性不可用,但您可以通过将其注入到您的配置类之一中来设置它。另请参阅远程文件信息SftpStreamingMessageSourcefalseSftpFileInfoLsEntrySftpFileInfo.getFileInfo()fileInfoJsonSftpStreamingMessageSource

从 5.1 版开始, 的泛型类型comparatorLsEntry. 以前,它是AbstractFileInfo<LsEntry>. 这是因为排序现在在处理的早期执行,在过滤和应用之前maxFetch

使用 Java 配置进行配置

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

@SpringBootApplication
public class SftpJavaApplication {

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

    @Bean
    @InboundChannelAdapter(channel = "stream")
    public MessageSource<InputStream> ftpMessageSource() {
        SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template());
        messageSource.setRemoteDirectory("sftpSource/");
        messageSource.setFilter(new AcceptAllFileListFilter<>());
        messageSource.setMaxFetchSize(1);
        return messageSource;
    }

    @Bean
    @Transformer(inputChannel = "stream", outputChannel = "data")
    public org.springframework.integration.transformer.Transformer transformer() {
        return new StreamTransformer("UTF-8");
    }

    @Bean
    public SftpRemoteFileTemplate template() {
        return new SftpRemoteFileTemplate(sftpSessionFactory());
    }

    @ServiceActivator(inputChannel = "data", adviceChain = "after")
    @Bean
    public MessageHandler handle() {
        return System.out::println;
    }

    @Bean
    public ExpressionEvaluatingRequestHandlerAdvice after() {
        ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
        advice.setOnSuccessExpression(
                "@template.remove(headers['file_remoteDirectory'] + headers['file_remoteFile'])");
        advice.setPropagateEvaluationFailures(true);
        return advice;
    }

}

请注意,在此示例中,转换器下游的消息处理程序具有在处理后删除远程文件的建议。

入站通道适配器:轮询多个服务器和目录

从版本 5.0.7 开始,RotatingServerAdvice可用;当配置为轮询建议时,入站适配器可以轮询多个服务器和目录。像往常一样配置建议并将其添加到轮询器的建议链中。ADelegatingSessionFactory用于选择服务器,请参阅委托会话工厂了解更多信息。建议配置由RotationPolicy.KeyDirectory对象列表组成。

例子
@Bean
public RotatingServerAdvice advice() {
    List<RotationPolicy.KeyDirectory> keyDirectories = new ArrayList<>();
    keyDirectories.add(new RotationPolicy.KeyDirectory("one", "foo"));
    keyDirectories.add(new RotationPolicy.KeyDirectory("one", "bar"));
    keyDirectories.add(new RotationPolicy.KeyDirectory("two", "baz"));
    keyDirectories.add(new RotationPolicy.KeyDirectory("two", "qux"));
    keyDirectories.add(new RotationPolicy.KeyDirectory("three", "fiz"));
    keyDirectories.add(new RotationPolicy.KeyDirectory("three", "buz"));
    return new RotatingServerAdvice(delegatingSf(), keyDirectories);
}

此建议将轮询foo服务器上的目录,one直到不存在新文件,然后移至目录bar,然后移至baz服务器上的目录two等。

可以使用fair构造函数 arg 修改此默认行为:

公平的
@Bean
public RotatingServerAdvice advice() {
    ...
    return new RotatingServerAdvice(delegatingSf(), keyDirectories, true);
}

在这种情况下,建议将移动到下一个服务器/目录,而不管之前的轮询是否返回了文件。

或者,您可以根据需要提供自己RotationPolicy的重新配置消息源:

政策
public interface RotationPolicy {

    void beforeReceive(MessageSource<?> source);

    void afterReceive(boolean messageReceived, MessageSource<?> source);

}

风俗
@Bean
public RotatingServerAdvice advice() {
    return new RotatingServerAdvice(myRotationPolicy());
}

local-filename-generator-expression属性(在localFilenameGeneratorExpression同步器上)现在可以包含#remoteDirectory变量。这允许从不同目录检索的文件下载到本地类似目录:

@Bean
public IntegrationFlow flow() {
    return IntegrationFlows.from(Sftp.inboundAdapter(sf())
                    .filter(new SftpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "rotate"))
                    .localDirectory(new File(tmpDir))
                    .localFilenameExpression("#remoteDirectory + T(java.io.File).separator + #root")
                    .remoteDirectory("."),
                e -> e.poller(Pollers.fixedDelay(1).advice(advice())))
            .channel(MessageChannels.queue("files"))
            .get();
}
TaskExecutor使用此建议时 不要在轮询器上配置 a ;有关详细信息,请参阅消息源的条件轮询器

入站通道适配器:控制远程文件获取

在配置入站通道适配器时,您应该考虑两个属性。 max-messages-per-poll,与所有轮询器一样,可用于限制每次轮询时发出的消息数量(如果准备好配置的值以上)。 max-fetch-size(从 5.0 版开始)可以限制一次从远程服务器检索的文件数。

以下场景假设起始状态是一个空的本地目录:

  • max-messages-per-poll=2and max-fetch-size=1:适配器获取一个文件,发出它,取出下一个文件,然后发出它。然后它会一直休眠到下一次投票。

  • max-messages-per-poll=2max-fetch-size=2):适配器获取两个文件,然后发出每个文件。

  • max-messages-per-poll=2and max-fetch-size=4:适配器最多获取 4 个文件(如果可用)并发出前两个(如果至少有两个)。接下来的两个文件将在下一次轮询时发出。

  • max-messages-per-poll=2且未max-fetch-size指定:适配器获取所有远程文件并发出前两个文件(如果至少有两个)。随后的文件在随后的轮询中发出(一次两个)。当所有文件都用完后,再次尝试远程获取,以获取任何新文件。

当您部署应用程序的多个实例时,我们建议设置一个小的max-fetch-size, 以避免一个实例“抓取”所有文件并饿死其他实例。

另一个用途max-fetch-size是当您想要停止获取远程文件但继续处理已经获取的文件时。在(以编程方式、通过 JMX 或通过控制总线)上设置maxFetchSize属性有效地阻止适配器获取更多文件,但让轮询器继续为先前获取的文件发出消息。如果属性更改时轮询器处于活动状态,则更改将在下一次轮询时生效。MessageSource

从 5.1 版开始,同步器可以提供一个Comparator<LsEntry>. 这在限制使用maxFetchSize.

SFTP 出站通道适配器

SFTP 出站通道适配器是一个特殊MessageHandler的连接到远程目录并为它接收到的每个文件启动文件传输作为传入的有效负载Message。它还支持文件的多种表示形式,因此您不受File对象的限制。与 FTP 出站适配器类似,SFTP 出站通道适配器支持以下负载:

  • java.io.File: 实际的文件对象

  • byte[]: 表示文件内容的字节数组

  • java.lang.String:表示文件内容的文本

  • java.io.InputStream:要传输到远程文件的数据流

  • org.springframework.core.io.Resource: 数据传输到远程文件的资源

以下示例显示如何配置 SFTP 出站通道适配器:

<int-sftp:outbound-channel-adapter id="sftpOutboundAdapter"
    session-factory="sftpSessionFactory"
    channel="inputChannel"
    charset="UTF-8"
    remote-file-separator="/"
    remote-directory="foo/bar"
    remote-filename-generator-expression="payload.getName() + '-mysuffix'"
    filename-generator="fileNameGenerator"
    use-temporary-filename="true"
    chmod="600"
    mode="REPLACE"/>

有关这些属性的更多详细信息,请参见架构

SpEL 和 SFTP 出站适配器

与 Spring Integration 中的许多其他组件一样,您可以在配置 SFTP 出站通道适配器时使用 Spring 表达式语言 (SpEL),方法是指定两个属性:remote-directory-expressionremote-filename-generator-expression如前所述)。表达式评估上下文将消息作为其根对象,它允许您使用可以根据消息中的数据(来自“有效负载”或“标题”)动态计算文件名或现有目录路径的表达式。在前面的示例中,我们使用表达式值定义remote-filename-generator-expression属性,该表达式值根据其原始名称计算文件名,同时还附加一个后缀:'-mysuffix'。

从版本 4.1 开始,您可以指定mode传输文件的时间。默认情况下,现有文件会被覆盖。模式由FileExistsMode枚举定义,包括以下值:

  • REPLACE(默认)

  • REPLACE_IF_MODIFIED

  • APPEND

  • APPEND_NO_FLUSH

  • IGNORE

  • FAIL

使用IGNOREFAIL,不传输文件。 FAIL导致抛出异常,同时IGNORE默默地忽略传输(尽管DEBUG会生成日志条目)。

4.3 版本引入了该chmod属性,您可以使用该属性在上传后更改远程文件权限。您可以使用传统的 Unix 八进制格式(例如,600只允许文件所有者读写)。使用 java 配置适配器时,可以使用setChmodOctal("600")setChmod(0600)

避免部分写入文件

处理文件传输时的常见问题之一是处理部分文件的可能性。文件可能在传输实际完成之前出现在文件系统中。

为了解决这个问题,Spring Integration SFTP 适配器使用一种通用算法,其中文件以临时名称传输,然后在完全传输后重命名。

默认情况下,正在传输的每个文件都出现在文件系统中,并带有一个附加后缀,默认情况下是.writing. 您可以通过设置temporary-file-suffix属性进行更改。

但是,在某些情况下您可能不想使用此技术(例如,如果服务器不允许重命名文件)。对于这种情况,您可以通过设置use-temporary-file-namefalse(默认为true)来禁用此功能。当此属性为false时,文件将写入其最终名称,并且消费应用程序需要一些其他机制来检测文件是否已完全上传,然后才能访问它。

使用 Java 配置进行配置

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

@SpringBootApplication
@IntegrationComponentScan
public class SftpJavaApplication {

    public static void main(String[] args) {
        ConfigurableApplicationContext context =
                    new SpringApplicationBuilder(SftpJavaApplication.class)
                        .web(false)
                        .run(args);
        MyGateway gateway = context.getBean(MyGateway.class);
        gateway.sendToSftp(new File("/foo/bar.txt"));
    }

    @Bean
    public SessionFactory<LsEntry> 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<LsEntry>(factory);
    }

    @Bean
    @ServiceActivator(inputChannel = "toSftpChannel")
    public MessageHandler handler() {
        SftpMessageHandler handler = new SftpMessageHandler(sftpSessionFactory());
        handler.setRemoteDirectoryExpressionString("headers['remote-target-dir']");
        handler.setFileNameGenerator(new FileNameGenerator() {

            @Override
            public String generateFileName(Message<?> message) {
                 return "handlerContent.test";
            }

        });
        return handler;
    }

    @MessagingGateway
    public interface MyGateway {

         @Gateway(requestChannel = "toSftpChannel")
         void sendToSftp(File file);

    }
}

使用 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 sftpOutboundFlow() {
        return IntegrationFlows.from("toSftpChannel")
            .handle(Sftp.outboundAdapter(this.sftpSessionFactory, FileExistsMode.FAIL)
                         .useTemporaryFileName(false)
                         .remoteDirectory("/foo")
            ).get();
    }

}

SFTP 出站网关

SFTP 出站网关提供了一组有限的命令,可让您与远程 SFTP 服务器交互:

  • ls(列出文件)

  • nlst(列出文件名)

  • get(检索文件)

  • mget(检索多个文件)

  • rm(删除文件)

  • mv(移动和重命名文件)

  • put(发送文件)

  • mput(发送多个文件)

使用ls命令

ls列出远程文件并支持以下选项:

  • -1: 检索文件名列表。默认是检索FileInfo对象列表

  • -a: 包括所有文件(包括以'.'开头的文件)

  • -f: 不对列表进行排序

  • -dirs: 包含目录(默认排除)

  • -links: 包括符号链接(默认不包括)

  • -R: 递归列出远程目录

此外,文件名过滤的提供方式与inbound-channel-adapter.

操作产生的消息有效负载ls是文件名列表或FileInfo对象列表(取决于您是否使用-1开关)。这些对象提供修改时间、权限等信息。

ls命令所作用的远程目录在file_remoteDirectory标头中提供。

使用递归选项 ( -R) 时,fileName包括任何子目录元素并表示文件的相对路径(相对于远程目录)。如果您使用该-dirs选项,则每个递归目录也将作为列表中的一个元素返回。在这种情况下,我们建议您不要使用该-1选项,因为您将无法区分文件和目录,而您可以在使用FileInfo对象时做到这一点。

使用nlst命令

版本 5 引入了对nlst命令的支持。

nlst列出远程文件名并且只支持一个选项:

  • -f: 不对列表进行排序

操作产生的消息有效负载nlst是文件名列表。

file_remoteDirectory头包含nlst命令所作用的远程目录。

SFTP 协议不提供列出名称的功能。该命令ls与带有-1选项的命令等效,为方便起见在此处添加。

使用get命令

get检索远程文件并支持以下选项:

  • -P:保留远程文件的时间戳。

  • -stream: 以流的形式检索远程文件。

  • -D: 传输成功后删除远程文件。如果忽略传输,远程文件不会被删除,因为FileExistsModeIGNORE和本地文件已经存在。

file_remoteDirectory头包含远程目录,file_remoteFile标头包含文件名。

get操作产生的消息有效负载是File表示检索到的文件的对象。如果您使用该-stream选项,则有效负载是 anInputStream而不是File. 对于文本文件,一个常见的用例是将此操作与文件拆分器流转换器结合使用。将远程文件作为流使用时,您有责任Session在使用流后关闭。为方便起见,在标题Session中提供了,并提供了方便的方法:closeableResourceIntegrationMessageHeaderAccessor

Closeable closeable = new IntegrationMessageHeaderAccessor(message).getCloseableResource();
if (closeable != null) {
    closeable.close();
}

框架组件,例如File SplitterStream Transformer,在数据传输后会自动关闭会话。

以下示例显示了如何将文件作为流使用:

<int-sftp:outbound-gateway session-factory="ftpSessionFactory"
                            request-channel="inboundGetStream"
                            command="get"
                            command-options="-stream"
                            expression="payload"
                            remote-directory="ftpTarget"
                            reply-channel="stream" />

<int-file:splitter input-channel="stream" output-channel="lines" />
如果您在自定义组件中使用输入流,则必须关闭Session. 您可以在自定义代码中执行此操作,也可以将消息副本路由到 aservice-activator并使用 SpEL,如以下示例所示:
<int:service-activator input-channel="closeSession"
    expression="headers['closeableResource'].close()" />

使用mget命令

mget基于模式检索多个远程文件并支持以下选项:

  • -P:保留远程文件的时间戳。

  • -R:递归检索整个目录树。

  • -x:如果没有文件与模式匹配,则抛出异常(否则,返回一个空列表)。

  • -D:成功传输后删除每个远程文件。如果忽略传输,则不会删除远程文件,因为FileExistsModeisIGNORE和本地文件已经存在。

操作产生的消息有效负载mget是一个List<File>对象(即一个List对象File,每个对象代表一个检索到的文件)。

从版本 5.0 开始,如果FileExistsModeIGNORE,则输出消息的有效负载不再包含由于文件已存在而未获取的文件。以前,该数组包含所有文件,包括那些已经存在的文件。

您使用的表达式确定远程路径应该产生一个结果,例如myfiles/fetches 下的完整树myfiles

从 5.0 版本开始,您可以使用 recursiveMGET结合FileExistsMode.REPLACE_IF_MODIFIED模式,定期在本地同步整个远程目录树。此模式将本地文件的最后修改时间戳设置为远程文件的时间戳,而不考虑-P(preserve timestamp) 选项。

使用递归时的注意事项 ( -R)

该模式被忽略并被*假定。默认情况下,检索整个远程树。但是,您可以通过提供FileListFilter. 您也可以通过这种方式过滤树中的目录。AFileListFilter可以通过引用filename-patternfilename-regex属性来提供。例如,检索以远程目录和子目录filename-regex="(subDir|.*1.txt)"结尾的所有文件。但是,我们在此说明之后描述了可用的替代方法。1.txtsubDir

如果过滤子目录,则不会执行对该子目录的额外遍历。

-dirs不允许该选项(递归mget使用递归ls获取目录树,目录本身不能包含在列表中)。

通常,您会在 中使用该#remoteDirectory变量,local-directory-expression以便在本地保留远程目录结构。

持久文件列表过滤器现在有一个布尔属性forRecursion。将此属性设置为true, 也设置alwaysAcceptDirectories,这意味着出站网关 (lsmget) 上的递归操作现在将始终遍历整个目录树。这是为了解决未检测到目录树深处更改的问题。此外,forRecursion=true使文件的完整路径用作元数据存储键;这解决了如果同名文件在不同目录中多次出现时过滤器无法正常工作的问题。重要提示:这意味着将无法为顶级目录下的文件找到持久元数据存储中的现有密钥。为此,该物业false默认; 这可能会在未来的版本中改变。

从 5.0 版开始,您可以通过将 设置为来配置SftpSimplePatternFileListFilterSftpRegexPatternFileListFilter以始终传递目录。这样做允许对简单模式进行递归,如以下示例所示:alwaysAcceptDirectortiestrue

<bean id="starDotTxtFilter"
            class="org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter">
    <constructor-arg value="*.txt" />
    <property name="alwaysAcceptDirectories" value="true" />
</bean>

<bean id="dotStarDotTxtFilter"
            class="org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter">
    <constructor-arg value="^.*\.txt$" />
    <property name="alwaysAcceptDirectories" value="true" />
</bean>

filter您可以使用网关上的属性提供这些过滤器之一。

使用put命令

put将文件发送到远程服务器。消息的有效负载可以是 a java.io.File、 abyte[]或 a String。A remote-filename-generator(或表达式)用于命名远程文件。其他可用属性包括remote-directory,temporary-remote-directory及其*-expression等价物:use-temporary-file-nameauto-create-directory. 有关更多信息,请参阅架构文档

操作产生的消息有效负载put是一个String包含传输后服务器上文件的完整路径的一个。

4.3 版本引入了该chmod属性,该属性会在上传后更改远程文件的权限。您可以使用传统的 Unix 八进制格式(例如,600只允许文件所有者读写)。使用 java 配置适配器时,可以使用setChmod(0600).

使用mput命令

mput向服务器发送多个文件并支持以下选项:

  • -R: 递归——发送目录和子目录中的所有文件(可能被过滤)

消息有效负载必须是表示本地目录的java.io.File(或)。从 5.1 版开始,也支持orString的集合。FileString

put支持与命令相同的属性。此外,您可以使用 、 、 或 之一过滤本地目录mput-patternmput-regexmput-filter文件mput-filter-expression。只要子目录本身通过过滤器,过滤器就可以使用递归。不通过过滤器的子目录不会被递归。

操作产生的消息有效负载mput是一个List<String>对象(即List传输产生的远程文件路径的一个)。

4.3 版本引入了该chmod属性,可让您在上传后更改远程文件权限。您可以使用传统的 Unix 八进制格式(例如,600只允许文件所有者读写)。使用 Java 配置适配器时,可以使用setChmodOctal("600")setChmod(0600)

使用rm命令

rm命令没有选项。

如果删除操作成功,则生成的消息有效负载为Boolean.TRUE. 否则,消息有效负载为Boolean.FALSE. 标file_remoteDirectory头包含远程目录,file_remoteFile标头包含文件名。

使用mv命令

mv命令没有选项。

属性定义“ expressionfrom”路径,rename-expression属性定义“to”路径。默认情况下,rename-expressionheaders['file_renameTo']. 此表达式的计算结果不得为 null 或空String。如有必要,将创建所需的任何远程目录。结果消息的有效负载是Boolean.TRUE。标file_remoteDirectory头包含原始远程目录,file_remoteFile标头包含文件名。标file_renameTo头包含新路径。

从版本 5.5.6 开始,为了方便,remoteDirectoryExpression可以在命令中使用。mv如果“from”文件不是完整的文件路径,则将结果remoteDirectoryExpression用作远程目录。这同样适用于“to”文件,例如,如果任务只是重命名某个目录中的远程文件。

附加命令信息

getandmget命令支持该属性local-filename-generator-expression。它定义了一个 SpEL 表达式以在传输过程中生成本地文件的名称。评估上下文的根对象是请求消息。该remoteFileName变量也可用。它对mget(例如:)特别有用local-filename-generator-expression="#remoteFileName.toUpperCase() + headers.foo"

getandmget命令支持该属性local-directory-expression。它定义了一个 SpEL 表达式以在传输过程中生成本地目录的名称。评估上下文的根对象是请求消息。该remoteDirectory变量也可用。它对 mget 尤其有用(例如:local-directory-expression="'/tmp/local/' + #remoteDirectory.toUpperCase() + headers.myheader")。该属性与该属性互斥local-directory

对于所有命令,网关的“表达式”属性保存命令作用的路径。对于该mget命令,表达式的计算结果可能为,表示检索所有文件、somedirectory/和其他以 结尾的值*

以下示例显示了为ls命令配置的网关:

<int-ftp:outbound-gateway id="gateway1"
        session-factory="ftpSessionFactory"
        request-channel="inbound1"
        command="ls"
        command-options="-1"
        expression="payload"
        reply-channel="toSplitter"/>

发送到toSplitter通道的消息的有效负载是一个String对象列表,每个对象都包含一个文件的名称。如果您省略command-options="-1",则有效负载将是FileInfo对象列表。您可以将选项提供为以空格分隔的列表(例如,command-options="-1 -dirs -links")。

从 4.2 版开始GETMGETPUT、 和MPUT命令支持FileExistsMode属性(mode使用命名空间支持时)。这会影响本地文件存在 ( GETand MGET) 或远程文件存在 ( PUTand MPUT) 时的行为。支持的模式是REPLACEAPPENDFAILIGNORE。为了向后兼容,PUTMPUT操作的默认模式是REPLACE. 对于GETMGET运算,默认为FAIL

使用 Java 配置进行配置

以下 Spring Boot 应用程序显示了如何使用 Java 配置出站网关的示例:

@SpringBootApplication
public class SftpJavaApplication {

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

    @Bean
    @ServiceActivator(inputChannel = "sftpChannel")
    public MessageHandler handler() {
        return new SftpOutboundGateway(ftpSessionFactory(), "ls", "'my_remote_dir/'");
    }

}

使用 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 SessionFactory<LsEntry> sftpSessionFactory() {
        DefaultSftpSessionFactory sf = new DefaultSftpSessionFactory();
        sf.setHost("localhost");
        sf.setPort(port);
        sf.setUsername("foo");
        sf.setPassword("foo");
        factory.setTestSession(true);
        return new CachingSessionFactory<LsEntry>(sf);
    }

    @Bean
    public QueueChannelSpec remoteFileOutputChannel() {
        return MessageChannels.queue();
    }

    @Bean
    public IntegrationFlow sftpMGetFlow() {
        return IntegrationFlows.from("sftpMgetInputChannel")
            .handle(Sftp.outboundGateway(sftpSessionFactory(),
                            AbstractRemoteFileOutboundGateway.Command.MGET, "payload")
                    .options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE)
                    .regexFileNameFilter("(subSftpSource|.*1.txt)")
                    .localDirectoryExpression("'myDir/' + #remoteDirectory")
                    .localFilenameExpression("#remoteFileName.replaceFirst('sftpSource', 'localTarget')"))
            .channel("remoteFileOutputChannel")
            .get();
    }

}

出站网关部分成功(mgetmput

在对多个文件执行操作(通过使用mgetand mput)时,可能会在传输一个或多个文件后的一段时间内发生异常。在这种情况下(从 4.2 版开始),PartialSuccessException会抛出 a。除了通常的MessagingException属性 (failedMessagecause),此异常还有两个附加属性:

  • partialResults: 传输成功的结果。

  • derivedInput:从请求消息生成的文件列表(例如要传输的本地文件mput)。

这些属性可让您确定哪些文件已成功传输,哪些未成功传输。

在递归的情况下mputPartialSuccessException可能有嵌套PartialSuccessException实例。

考虑以下目录结构:

root/
|- file1.txt
|- subdir/
   | - file2.txt
   | - file3.txt
|- zoo.txt

如果异常发生在file3.txt,网关抛出的PartialSuccessExceptionderivedInputof file1.txtsubdirzoo.txtpartialResultsfile1.txt它是cause另一个PartialSuccessExceptionderivedInputoffile2.txtfile3.txtof partialResultsfile2.txt

SFTP/JSCH 日志记录

由于我们使用 JSch 库来提供 SFTP 支持,因此您有时可能需要来自 JSch API 本身的更多信息,尤其是在某些东西无法正常工作的情况下(例如身份验证异常)。不幸的是,JSch 不使用commons-logging,而是依赖于其com.jcraft.jsch.Logger接口的自定义实现。从 Spring Integration 2.0.1 开始,我们已经实现了这个接口。所以现在,要启用 JSch 日志记录,您可以按照通常的方式配置您的记录器。例如,以下示例是使用 Log4J 的记录器的有效配置:

log4j.category.com.jcraft.jsch=DEBUG

消息会话回调

从 Spring Integration 版本 4.2 开始,您可以使用MessageSessionCallback<F, T>带有<int-sftp:outbound-gateway/>( ) 的实现来对带有上下文SftpOutboundGateway的 执行任何操作。您可以将它用于任何非标准或低级 SFTP 操作(或多个),例如允许从集成流定义或功能接口 (lambda) 实现注入进行访问。以下示例使用 lambda:Session<LsEntry>requestMessage

@Bean
@ServiceActivator(inputChannel = "sftpChannel")
public MessageHandler sftpOutboundGateway(SessionFactory<ChannelSftp.LsEntry> sessionFactory) {
    return new SftpOutboundGateway(sessionFactory,
         (session, requestMessage) -> session.list(requestMessage.getPayload()));
}

另一个示例可能是对正在发送或检索的文件数据进行预处理或后处理。

使用 XML 配置时,它<int-sftp:outbound-gateway/>提供了一个session-callback属性,可以让您指定MessageSessionCallbackbean 名称。

与和属性session-callback互斥。使用 Java 进行配置时,该类提供了不同的构造函数。 commandexpressionSftpOutboundGateway

Apache Mina SFTP 服务器事件

在 5.2 版中添加的ApacheMinaSftpEventListener,侦听某些 Apache Mina SFTP 服务器事件并将它们发布为ApplicationEvent可以被任何ApplicationListenerbean、@EventListenerbean 方法或Event Inbound Channel Adapter接收的 s 。

目前支持的事件有:

  • SessionOpenedEvent- 客户端会话已打开

  • DirectoryCreatedEvent- 创建了一个目录

  • FileWrittenEvent- 一个文件被写入

  • PathMovedEvent- 文件或目录被重命名

  • PathRemovedEvent- 删除了文件或目录

  • SessionClosedEvent- 客户端已断开连接

这些中的每一个都是ApacheMinaSftpEvent; 您可以配置单个侦听器来接收所有事件类型。每个事件的source属性是一个ServerSession,可以从中获取客户端地址等信息;在抽象事件上提供了一种方便的getSession()方法。

要使用侦听器(必须是 Spring bean)配置服务器,只需将其添加到SftpSubsystemFactory

server = SshServer.setUpDefaultServer();
...
SftpSubsystemFactory sftpFactory = new SftpSubsystemFactory();
sftpFactory.addSftpEventListener(apacheMinaSftpEventListenerBean);
...

要使用 Spring Integration 事件适配器使用这些事件:

@Bean
public ApplicationEventListeningMessageProducer eventsAdapter() {
    ApplicationEventListeningMessageProducer producer =
        new ApplicationEventListeningMessageProducer();
    producer.setEventTypes(ApacheMinaSftpEvent.class);
    producer.setOutputChannel(eventChannel());
    return producer;
}

远程文件信息

从版本 5.2 开始,SftpStreamingMessageSourceSFTP 流入站通道适配器)、SftpInboundFileSynchronizingMessageSourceSFTP 入站通道适配器SftpOutboundGateway)和(SFTP 出站网关)的“读取”命令在消息中提供额外的标头,以生成有关远程文件的信息:

  • FileHeaders.REMOTE_HOST_PORT- 在文件传输操作期间远程会话已连接到的主机:端口对;

  • FileHeaders.REMOTE_DIRECTORY- 已执行操作的远程目录;

  • FileHeaders.REMOTE_FILE- 远程文件名;仅适用于单个文件操作。

由于SftpInboundFileSynchronizingMessageSource不生成针对远程文件的消息,而是使用本地副本,因此在同步操作期间,AbstractInboundFileSynchronizer将有关远程文件的信息存储MetadataStore在 URI 样式 ( ) 中(可以在外部配置) 。protocol://host:port/remoteDirectory#remoteFileName此元数据由SftpInboundFileSynchronizingMessageSource轮询本地文件时检索。当本地文件被删除时,建议删除其元数据条目。为此目的提供AbstractInboundFileSynchronizer回调。removeRemoteFileMetadata()此外setMetadataStorePrefix(),元数据键中还有一个要使用的键。当相同的实例在这些组件之间共享时,建议将此前缀与MetadataStore基于- 的实现中使用的前缀不同,以避免条目覆盖,因为过滤器和FileListFilterMetadataStoreAbstractInboundFileSynchronizer元数据条目键使用相同的本地文件名。


1. see XML Configuration