如何通过Spring Integration从SFTP下载最新修改的ZIP文件？

开发了一个从SFTP服务器下载文件的应用，需求为：

• 仅下载ZIP格式文件
• 仅在文件被修改后才重新下载
  目前已实现SftpInboundFileSynchronizer和多个InboundChannelAdapter，但出现同一文件重复下载的问题。尝试过SftpPersistentAcceptOnceFileListFilter(ms, "gtfs_")、SftpSimplePatternFileListFilter("*.zip")等过滤器，但未解决问题。相关代码如下：

public static final String SYNCHRONIZER_BEAN_NAME = "synchorinzer-bean-name";
    @Bean(SYNCHRONIZER_BEAN_NAME)
    public SftpInboundFileSynchronizer synchronizer(
            SessionFactory<SftpClient.DirEntry> sf,
            PropertiesPersistingMetadataStore ms,
            AppProps cfg) {
        SftpInboundFileSynchronizer sync = new SftpInboundFileSynchronizer(sf);
        sync.setDeleteRemoteFiles(false);
        sync.setRemoteDirectory(cfg.getFtpRemoteDirectory());
        sync.setPreserveTimestamp(true);
        // sync.setFilter(); ????
        return sync;
    }


    public static final String GIPUZKOANA_OUT_CHANNEL_NAME = "GIPUZKOANA_OUT_CHANNEL";
    public static final String GIPUZKOANA_SYNCHRONIZER_BEAN_NAME = "GIPUZKOANA_FILE_SYNCHRONIZER_BEAN";
    @Bean(GIPUZKOANA_SYNCHRONIZER_BEAN_NAME)
    @InboundChannelAdapter(channel = GIPUZKOANA_OUT_CHANNEL_NAME)
    public MessageSource<File> gipuzkoanaMessageSource(
            @Qualifier(SYNCHRONIZER_BEAN_NAME) SftpInboundFileSynchronizer sync,
            AppProps cfg) {
        SftpInboundFileSynchronizingMessageSource source = new SftpInboundFileSynchronizingMessageSource(sync);
        source.setLocalDirectory(cfg.getGtfsLocalDirSyncGtfs());
        source.setAutoCreateLocalDirectory(true);
        source.setMaxFetchSize(1);
        source.setLoggingEnabled(true);
        source.setLocalFilter(files -> Lists.newArrayList(files)
                .stream()
                .filter(f -> f.getName().equalsIgnoreCase(cfg.getGtfsGipuzkoana()))
                .collect(Collectors.toList()));
        return source;
    }
    // ...

问题核心在于远程过滤器未正确配置，同时本地过滤器的写法绕过了内置的重复处理检查。按以下步骤修改：

1. 配置组合远程过滤器 #

在SftpInboundFileSynchronizer中同时设置格式过滤和持久化修改时间检查的过滤器，确保只下载更新过的ZIP文件：

@Bean(SYNCHRONIZER_BEAN_NAME)
public SftpInboundFileSynchronizer synchronizer(
        SessionFactory<SftpClient.DirEntry> sf,
        PropertiesPersistingMetadataStore ms,
        AppProps cfg) {
    SftpInboundFileSynchronizer sync = new SftpInboundFileSynchronizer(sf);
    sync.setDeleteRemoteFiles(false);
    sync.setRemoteDirectory(cfg.getFtpRemoteDirectory());
    sync.setPreserveTimestamp(true);
    
    // 组合两个过滤器：先筛选ZIP文件，再检查是否已处理或更新
    CompositeFileListFilter<SftpClient.DirEntry> remoteFilter = new CompositeFileListFilter<>();
    remoteFilter.addFilter(new SftpSimplePatternFileListFilter("*.zip"));
    // 前缀"gtfs_"用于区分不同适配器的元数据记录，避免冲突
    remoteFilter.addFilter(new SftpPersistentAcceptOnceFileListFilter(ms, "gtfs_"));
    sync.setFilter(remoteFilter);
    
    return sync;
}

2. 优化本地过滤器 #

当前本地过滤器使用Lambda写法，会跳过Spring Integration内置的本地文件去重逻辑，改用组合过滤器实现文件名匹配+本地文件去重：

@Bean(GIPUZKOANA_SYNCHRONIZER_BEAN_NAME)
@InboundChannelAdapter(channel = GIPUZKOANA_OUT_CHANNEL_NAME)
public MessageSource<File> gipuzkoanaMessageSource(
        @Qualifier(SYNCHRONIZER_BEAN_NAME) SftpInboundFileSynchronizer sync,
        AppProps cfg) {
    SftpInboundFileSynchronizingMessageSource source = new SftpInboundFileSynchronizingMessageSource(sync);
    source.setLocalDirectory(cfg.getGtfsLocalDirSyncGtfs());
    source.setAutoCreateLocalDirectory(true);
    source.setMaxFetchSize(1);
    source.setLoggingEnabled(true);
    
    // 组合本地过滤器：匹配指定文件名 + 避免本地重复处理
    CompositeFileListFilter<File> localFilter = new CompositeFileListFilter<>();
    localFilter.addFilter(new SimplePatternFileListFilter(cfg.getGtfsGipuzkoana()));
    localFilter.addFilter(new AcceptOnceFileListFilter<>());
    source.setLocalFilter(localFilter);
    
    return source;
}

3. 关键注意事项 #

• 确保PropertiesPersistingMetadataStore配置了持久化路径，否则应用重启后会丢失已处理文件的记录，导致重新下载所有文件
• 每个InboundChannelAdapter对应的SftpPersistentAcceptOnceFileListFilter前缀要唯一，避免不同任务的元数据互相干扰
• 必须保留setPreserveTimestamp(true)，该配置会让本地文件的修改时间与远程保持一致，确保过滤器能正确识别文件是否更新

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