集群环境下SFTP流式入站通道适配器实现问题咨询

我来帮你梳理下问题的核心和解决方案——你遇到的问题本质是PropertiesMetadataStore不支持集群共享：它是基于本地文件系统存储元数据的，每个节点都会维护自己独立的处理记录，完全不知道其他节点已经处理过哪些文件，自然会出现重复处理、文件找不到（比如节点A先处理并删除/移动了文件，节点B轮询时就找不到了）的情况。

下面是针对集群场景的可靠解决方案：

1. 替换为分布式MetadataStore #

必须改用支持跨节点共享元数据的MetadataStore实现，常用的有三种：

• RedisMetadataStore：基于Redis，适合大多数分布式场景，配置简单性能好
• ZookeeperMetadataStore：基于ZooKeeper，适合需要强一致性锁的场景
• JdbcMetadataStore：基于关系型数据库，适合已有数据库集群的环境

这里以Redis为例，给出核心配置代码：

@Bean
public MetadataStore metadataStore(RedisConnectionFactory connectionFactory) {
    RedisMetadataStore metadataStore = new RedisMetadataStore(connectionFactory);
    metadataStore.setPrefix("sftp-streaming-task-"); // 给当前SFTP任务设置唯一前缀，避免和其他任务冲突
    return metadataStore;
}

// 配置SFTP流式入站适配器时关联这个分布式MetadataStore
@Bean
public SftpStreamingInboundChannelAdapter sftpStreamingAdapter(SftpRemoteFileTemplate sftpTemplate, MetadataStore metadataStore) {
    SftpStreamingInboundChannelAdapter adapter = new SftpStreamingInboundChannelAdapter(sftpTemplate);
    adapter.setRemoteDirectory("/remote/sftp/target");
    // 使用分布式持久化过滤器，确保已处理文件不会被重复轮询
    adapter.setFilter(new SftpPersistentAcceptOnceFileListFilter(metadataStore, "sftp-file-"));
    adapter.setMaxFetchSize(1); // 集群环境建议每次轮询只取1个文件，减少节点竞争
    return adapter;
}

2. 实现SFTP文件的原子性处理 #

为了避免节点间的文件竞争，处理文件前要先做原子性的移动操作，确保同一时间只有一个节点能处理该文件：

• 处理前将文件从源目录原子移动到processing临时目录
• 处理完成后再移动到processed目录或直接删除
• 处理失败则移到error目录，方便后续排查

示例业务处理代码：

@Service
public class SftpFileHandler {
    private final SftpRemoteFileTemplate sftpTemplate;

    public SftpFileHandler(SftpRemoteFileTemplate sftpTemplate) {
        this.sftpTemplate = sftpTemplate;
    }

    public void processStream(InputStream inputStream, String remoteFilePath) {
        String processingPath = remoteFilePath.replace("/remote/sftp/target", "/remote/sftp/processing");
        String processedPath = remoteFilePath.replace("/remote/sftp/target", "/remote/sftp/processed");
        String errorPath = remoteFilePath.replace("/remote/sftp/target", "/remote/sftp/error");

        try {
            // 原子移动到processing目录，确保其他节点看不到原文件
            sftpTemplate.rename(remoteFilePath, processingPath);
            
            // 流式处理文件内容（这里替换成你的业务逻辑）
            parseAndPersistStream(inputStream);
            
            // 处理完成后移动到processed目录
            sftpTemplate.rename(processingPath, processedPath);
        } catch (Exception e) {
            // 处理失败，移到error目录
            sftpTemplate.rename(processingPath, errorPath);
            throw new RuntimeException("Failed to process SFTP file: " + remoteFilePath, e);
        }
    }

    private void parseAndPersistStream(InputStream inputStream) {
        // 你的业务处理：比如解析CSV、写入数据库等
    }
}

3. 给轮询器加分布式锁 #

除了共享元数据，还可以给轮询器加上分布式锁，确保同一时间只有一个节点在轮询SFTP服务器，从根源减少竞争：

@Bean
public LockRegistry lockRegistry(RedisConnectionFactory connectionFactory) {
    return new RedisLockRegistry(connectionFactory, "sftp-polling-lock-group");
}

@Bean
public PollerMetadata sftpPoller(LockRegistry lockRegistry) {
    PeriodicTrigger trigger = new PeriodicTrigger(60000); // 每分钟轮询一次，可根据业务调整
    trigger.setFixedRate(true);

    return Pollers.trigger(trigger)
            .lock(lockRegistry.obtain("sftp-poll-lock")) // 给轮询器加全局锁
            .get();
}

// 整合到集成流中
@Bean
public IntegrationFlow sftpIntegrationFlow(SftpStreamingInboundChannelAdapter adapter, PollerMetadata sftpPoller) {
    return IntegrationFlows.from(adapter, spec -> spec.poller(sftpPoller))
            .handle(SftpFileHandler.class, "processStream")
            .get();
}

4. 排查常见坑点 #

• 确保所有集群节点的MetadataStore前缀完全一致，否则节点间的元数据无法互通
• 检查SFTP账号权限：所有节点使用的SFTP账号必须拥有源目录、processing/processed/error目录的读写权限
• 若使用JdbcMetadataStore，要确保所有节点连接同一个数据库实例，并且数据库表已正确初始化（Spring会自动创建表，但需确保权限足够）
• 流式处理逻辑要尽量轻量化，避免长时间占用文件流，减少节点间的竞争窗口

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