Spring Integration动态多SFTP/FTP会话及集群等技术问题咨询

嘿，针对你提到的两个核心问题，我结合实际项目经验给你梳理下可行的方案：

一、动态添加SFTP/FTP任务 #

完全可以实现！Spring Integration提供了IntegrationFlowContext来动态创建和管理集成流，刚好适配从数据库读取连接信息、批量创建轮询任务的场景。

具体步骤大概是这样：

• 从数据库中读取所有SFTP/FTP的连接配置（主机、端口、用户名、密码、路径等）
• 遍历每个配置，用SftpInboundChannelAdapterSpec（SFTP）或FtpInboundChannelAdapterSpec（FTP）构建对应的入站适配器
• 通过IntegrationFlowContext将构建好的集成流注册到Spring上下文，还可以给每个流分配唯一ID，方便后续更新或移除

举个简化的代码示例：

@Autowired
private IntegrationFlowContext flowContext;

public void registerSftpTask(SftpConfig config) {
    // 针对每个服务器动态创建专属SessionFactory
    DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory();
    factory.setHost(config.getHost());
    factory.setPort(config.getPort());
    factory.setUser(config.getUsername());
    factory.setPassword(config.getPassword());

    IntegrationFlow flow = IntegrationFlow.from(Sftp.inboundAdapter(factory)
                    .remoteDirectory(config.getRemoteDir())
                    .localDirectory(new File(config.getLocalDir()))
                    .deleteRemoteFiles(false),
            e -> e.poller(Pollers.fixedDelay(config.getPollIntervalSeconds(), TimeUnit.SECONDS)))
            .handle(message -> {
                // 自定义文件处理逻辑
                System.out.println("处理文件：" + message.getPayload());
            })
            .get();

    // 用配置ID作为流的唯一标识，方便后续管理
    flowContext.registration(flow).id("sftp-flow-" + config.getId()).register();
}

如果后续需要更新或删除某个任务，直接调用flowContext.remove("sftp-flow-" + config.getId())即可。

二、集群部署下的任务控制 #

Spring Integration本身支持集群部署，但要注意避免同一任务被多个节点重复执行，核心是通过分布式锁来控制轮询任务的执行权。

常用的方案是：

• 使用Spring Integration的LockRegistry接口，结合Redis/Zookeeper实现分布式锁（比如RedisLockRegistry）
• 在Poller中配置锁，确保同一时间只有一个节点能执行轮询操作

示例代码片段：

// 配置Redis分布式锁注册表
@Bean
public LockRegistry redisLockRegistry(RedisConnectionFactory connectionFactory) {
    return new RedisLockRegistry(connectionFactory, "sftp-poller-locks");
}

// 构建流时给Poller绑定专属锁
IntegrationFlow flow = IntegrationFlow.from(Sftp.inboundAdapter(factory)
                .remoteDirectory(config.getRemoteDir())
                .localDirectory(new File(config.getLocalDir())),
        e -> e.poller(Pollers.fixedDelay(config.getPollIntervalSeconds(), TimeUnit.SECONDS)
                .lock(redisLockRegistry.obtain("lock-" + config.getId())))) // 每个任务对应唯一锁
        .handle(...)
        .get();

这样集群中的多个节点会竞争同一把锁，只有拿到锁的节点会执行轮询拉取操作，避免重复处理文件。

关于你提到的多SFTP方案无效的问题 #

之前找到的方案没生效，大概率是没正确使用IntegrationFlowContext来动态管理集成流，或者没处理好SessionFactory的动态创建（比如复用了同一个工厂导致配置冲突）。按照上面的动态注册+分布式锁的思路，应该能解决多SFTP服务器轮询的问题。

内容的提问来源于stack exchange，提问作者Philipp K.