如何配置Spring Integration Poller实现SFTP文件周期性下载

我希望周期性调用downloadFiles()方法，从远程服务器的子目录下载文件，并保存到本地同名文件夹中。原本以为<int:poller fixed-rate="1000"/>可以实现这一功能，但显然我没完全理解Poller的概念。

代码文件 #

SftpOutboundGateway-context.xml #

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

    <import resource="web.xml"/>

    <int-sftp:outbound-gateway id="gateway1" 
                              auto-startup="true"
                              session-factory="sftpSessionFactory"
                              request-channel="receiveChannel"
                              command="mget"
                              local-directory-expression="'C:/temp/cishare/' + headers['register'] "
                              auto-create-local-directory="true"
                              rename-expression=""
                              mode="IGNORE"
                              expression="'echo_localhost/echo_localhost/' + payload"
                              reply-channel="loggingChannel"
    >
        <int:poller fixed-rate="20"/>
    </int-sftp:outbound-gateway>

    <int:channel id="receiveChannel">
        <int:queue capacity="25"/>
    </int:channel>

    <int:logging-channel-adapter id="loggingChannel" log-full-message="true" logger-name="tapInbound"
                                 level="INFO" />
    
</beans>

SftpTransfer.java #

import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;

import java.util.ArrayList;

public class SftpTransfer {

    public void downloadFiles(MessageChannel receiveChannel) {
        ArrayList<String> directories = new ArrayList<>();
        directories.add("IN/");
        directories.add("OUT/");

        for (String directory : directories) {
            Message<String> requestMessage = MessageBuilder.withPayload(directory)
                    .setHeader("register", directory)
                    .build();

            receiveChannel.send(requestMessage);
        }
    }
}

App.java #

import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.messaging.MessageChannel;

public class App {

    public static void main(String[] args) {
        ConfigurableApplicationContext ctx = new ClassPathXmlApplicationContext(
                "classpath:SftpOutboundGateway-context.xml");

        MessageChannel receiveChannel = (MessageChannel) ctx.getBean("receiveChannel");

        SftpTransfer sftpTransfer = new SftpTransfer();
        sftpTransfer.downloadFiles(receiveChannel);

    }

}

问题分析与解决方案 #

你当前配置中的<int:poller>是绑定在SFTP Outbound Gateway上的，它的作用是轮询receiveChannel队列中的消息，触发网关执行下载操作，而非周期性调用downloadFiles()方法。你的downloadFiles()仅在应用启动时被调用一次，后续不会自动重复执行，因此无法实现周期性下载。

要实现周期性触发下载逻辑，需要将定时规则绑定到downloadFiles()方法的调用上，以下是两种可行方案：

方案1：XML配置定时入站通道适配器 #

在XML配置中添加定时触发的入站适配器，周期性调用downloadFiles()：

<!-- 注册SftpTransfer实例 -->
<bean id="sftpTransfer" class="com.yourpackage.SftpTransfer"/>

<!-- 定时触发下载方法 -->
<int:inbound-channel-adapter channel="triggerChannel"
                             expression="@sftpTransfer.downloadFiles(receiveChannel)">
    <int:poller fixed-rate="1000"/> <!-- 每隔1秒触发一次，可按需调整 -->
</int:inbound-channel-adapter>

<int:channel id="triggerChannel"/>

这里的<int:poller>作用是每隔指定时间执行一次expression中的方法调用，从而周期性触发下载逻辑。

方案2：使用@Scheduled注解 #

修改SftpTransfer类，通过Spring Task的注解实现定时调用：

import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.ArrayList;

public class SftpTransfer {

    @Autowired
    private MessageChannel receiveChannel;

    @Scheduled(fixedRate = 1000) // 每隔1秒执行一次
    public void downloadFiles() {
        ArrayList<String> directories = new ArrayList<>();
        directories.add("IN/");
        directories.add("OUT/");

        for (String directory : directories) {
            Message<String> requestMessage = MessageBuilder.withPayload(directory)
                    .setHeader("register", directory)
                    .build();

            receiveChannel.send(requestMessage);
        }
    }
}

同时在XML配置中开启定时任务支持：

<task:annotation-driven/>
<bean id="sftpTransfer" class="com.yourpackage.SftpTransfer"/>

关键注意事项 #

1. 移除SFTP Outbound Gateway上的<int:poller>：receiveChannel是队列通道，网关会自动处理队列中的消息，额外配置Poller会造成不必要的资源消耗。
2. 调整fixed-rate值到合理周期（比如300000ms即5分钟），避免过于频繁的SFTP连接请求。
3. 确保项目已引入Spring Integration和Spring Task的相关依赖。

内容的提问来源于stack exchange，提问作者terrinizer