SFTP Spring集成中的文件轮询和过滤问题

Question

我正在尝试通过SFTP Spring集成拉取远程文件并将其转储到本地驱动器中，并启动Spring批处理作业（步骤1读取、处理和写入），接下来是步骤2，其中一个Tasklet将下载到本地另一个位置的本地文件存档并删除下载的文件。

假设下载的文件是“data.service”、“data.maintenance”和“data.transaction”。我只处理“data.service”，但所有3个都已存档并删除。

假设一组新的文件将在远程位置被覆盖，我希望以上步骤以特定的时间间隔发生。

<bean id="acceptAllFileListFilter" class="org.springframework.integration.file.filters.AcceptAllFileListFilter"/>
<bean id="acceptOnceFileListFilter" class="org.springframework.integration.file.filters.AcceptOnceFileListFilter"/>

conf

<int-sftp:inbound-channel-adapter id="sftpInbondAdapter"
        channel="inboundFileChannel"
        session-factory="sftpSessionFactory"
        local-directory="file:${inbound.local.directory}"
        remote-directory="${inbound.remote.directory}"
        auto-create-local-directory="true"
        delete-remote-files="false"
        preserve-timestamp="true"
        filename-pattern="*.*"
        local-filter="acceptAllFileListFilter" >
    <int:poller max-messages-per-poll="-1" fixed-rate="60000" />
</int-sftp:inbound-channel-adapter>

现在发生的是...上面的操作触发了这个序列，并在一个无休止的循环中运行，而不需要从远程获取新文件。详情如下：

我从一个空的本地文件夹开始，远程服务器有所有3个文件[开始程序]

null

        local-filter="acceptOnceFileListFilter" >

null

对于不同职位的建议，我有一个独特的工作参数，如下所示：

@Transformer
public JobLaunchRequest toRequest(Message<File> message) {
    JobParametersBuilder jobParametersBuilder = new JobParametersBuilder();
    jobParametersBuilder.addString(fileParameterName, message.getPayload().getAbsolutePath());
    jobParametersBuilder.addLong("currentTime", new Long(System.currentTimeMillis()));
    return new JobLaunchRequest(job, jobParametersBuilder.toJobParameters());
}

尝试了不同的过滤器组合，没有一个达到我的目的。任何样本都会有很大的帮助，从上周开始就在努力解决这个问题。

以下是我根据建议所做的修改：

<bean id="compositeFilter" class="org.springframework.integration.file.filters.CompositeFileListFilter">
    <constructor-arg>
        <list>
            <bean class="org.springframework.integration.file.filters.FileSystemPersistentAcceptOnceFileListFilter"/>
            <bean class="org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter">
            </bean>                     
        </list>
    </constructor-arg>
</bean>

<int-sftp:inbound-channel-adapter id="sftpInbondAdapter"
        channel="inboundFileChannel"
        session-factory="sftpSessionFactory"
        local-directory="file:${inbound.local.directory}"
        remote-directory="${inbound.remote.directory}"
        auto-create-local-directory="true"
        delete-remote-files="false"  
        filter="compositeFilter">
    <int:poller max-messages-per-poll="-1" fixed-rate="600000" />
</int-sftp:inbound-channel-adapter>

<int-sftp:inbound-channel-adapter id="sftpInbondAdapter"
        channel="inboundFileChannel"
        session-factory="sftpSessionFactory"
        local-directory="file:${inbound.local.directory}"
        remote-directory="file:${inbound.remote.directory}"
        auto-create-local-directory="false"
        delete-remote-files="false"  
        filter="compositeFilter"
        >
    <int:poller max-messages-per-poll="-1" fixed-rate="60000" />
</int-sftp:inbound-channel-adapter>

<bean id="compositeFilter" class="org.springframework.integration.file.filters.CompositeFileListFilter">
    <constructor-arg>
        <list>
            <bean class="org.springframework.integration.file.filters.FileSystemPersistentAcceptOnceFileListFilter">
                <constructor-arg name="store" ref="metadataStore"/>
                <constructor-arg value="*.*"/>
            </bean>
            <bean class="org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter">
                <constructor-arg name="store" ref="metadataStore"/>
                <constructor-arg value="*.*"/>
            </bean>
        </list>
    </constructor-arg>
</bean>

<bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
    <property name="port" value="7379"/>
</bean>

<bean name="metadataStore" class="org.springframework.integration.redis.metadata.RedisMetadataStore">
    <constructor-arg name="connectionFactory" ref="redisConnectionFactory"/>
</bean>

2014-07-17 22:27:14,199[task-scheduler-1]调试org.SpringFramework.Integration.channel.PublishSubscribeChannel-postSend(SENT=TRUE)对通道“错误通道”，消息：[Payload MessagingException Content=org.SpringFramework.Messaging.MessagingException：将html" target="_blank">远程目录同步到本地目录时出现问题][Headers={ID=84CCCECF-20EC-F67B-3F74-A938BF9ABF3D,TimeStamp=1405661234197}]

删除FileSystemPersistentAcceptonCeFileListFilter筛选器离开“SFTPPersistentAcceptonCeFileListFilter”后更新

2014-07-18 09:29:22,069[task-scheduler-3]调试org.SpringFramework.Integration.channel.PublishSubscribeChannel-postSend(SENT=TRUE)对通道“错误通道”，消息：[Payload MessagingException Content=org.SpringFramework.Messaging.MessagingException：将远程目录同步到本地目录时出现问题][Headers={ID=55304E38-6F82-8350-E763-0F5BCB507788,TimeStamp=1405700962068}]

Answer 1

可能是这样的情况：如果丢弃在服务器上的新文件具有相同的名称（在第一次迭代中下载），那么spring integration将不会再次尝试下载这些文件。“AcceptonCEFileListFilter”似乎会限制sftp集成组件再次下载同一文件。

更新：这里是一些示例代码，它每10秒轮询一次远程目录，而不引入任何过滤器（我不知道为什么您需要它们，因为您想下载文件）。如果您不想更改文件名，您必须确保的唯一一件事是，当下一个文件被丢弃在远程服务器上时，您下载的本地文件需要删除。我验证了它，对我来说工作得很好。

<?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-ftp="http://www.springframework.org/schema/integration/ftp"
   xsi:schemaLocation="http://www.springframework.org/schema/beans      http://www.springframework.org/schema/beans/spring-beans.xsd
   http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd

   http://www.springframework.org/schema/integration/ftp http://www.springframework.org/schema/integration/ftp/spring-integration-ftp.xsd">


<int:channel id="incomingChannel"/>

<bean id="ftpClientFactory" class="org.springframework.integration.ftp.session.DefaultFtpSessionFactory">
    <property name="host" value="host"/>
    <property name="username" value="username"/>
    <property name="password" value="password"/>
</bean>

<!-- poll every 10 secs without worrying about file names-->

<int-ftp:inbound-channel-adapter id="triggerFtpInBound"
                                 channel="incomingChannel"
                                 auto-create-local-directory="true"
                                 local-directory="C:\home\temp\ftp"
                                 remote-directory="/export/home/rbaljinder/ftp-test"
                                 filename-pattern="*.*"
                                 session-factory="ftpClientFactory">
    <int:poller cron="1/10 * * * * *" max-messages-per-poll="1"/>
</int-ftp:inbound-channel-adapter>

更新：现在似乎起作用了。added local-filter=“AcceptAllFileListFilter”。

 <int-ftp:inbound-channel-adapter id="triggerFtpInBound"
                                 channel="incomingChannel"
                                 auto-create-local-directory="true"
                                 local-directory="C:\home\temp\ftp"
                                 remote-directory="/export/home/cwk2/ftp-test"
                                 filename-pattern="*.*"
                                 session-factory="ftpClientFactory"
                                 local-filter="acceptAllFileListFilter">
    <int:poller cron="1/10 * * * * *" max-messages-per-poll="1"/>
</int-ftp:inbound-channel-adapter>
<bean id="acceptAllFileListFilter"
      class="org.springframework.integration.file.filters.AcceptAllFileListFilter"/>
<int:service-activator id="jobServiceActivator"
                       input-channel="incomingChannel"
                       ref="triggerJobLauncher"
                       method="launch"/>

@Component("triggerJobLauncher")
public static class TriggerJobLauncher {

    @Autowired
    JobLauncher jobLauncher;

    public void launch(File file) throws Exception {
        System.out.println("test:" + file);
    }

}