Spring Integration 远程持久文件列表过滤器详解

概述

远程持久文件列表过滤器(Remote Persistent File List Filters)是 Spring Integration 文件处理的核心组件，用于避免重复处理远程文件（如 FTP/SFTP 服务器上的文件）。本教程将循序渐进地解释其工作原理和使用方法。

TIP

在分布式系统中，这些过滤器能确保集群节点不会重复处理相同文件，是构建可靠文件处理系统的关键组件。

核心概念解析

1. 过滤器的作用机制

2. 版本演进对比

特性	5.2 之前版本	5.2+ 版本
文件标记时机	获取前标记为已处理	获取前标记为"处理中"
故障恢复	可能丢失多个文件	最多丢失一个文件
单文件过滤	不支持	支持
递归处理	目录处理有限	支持完整递归

关键API与配置

1. 基础过滤器接口

interface FileListFilter<F> {
    // 检查单个文件是否应该被处理
    fun accept(file: F): Boolean

    // 是否支持单文件过滤
    fun supportsSingleFileFiltering(): Boolean
}

IMPORTANT

自定义过滤器必须实现 accept() 方法，否则适配器会回退到旧版本行为！

2. 集群环境配置示例

@Configuration
class RemoteFileConfig {

    // 使用Redis作为共享元数据存储
    @Bean
    fun sharedMetadataStore(): MetadataStore {
        return RedisMetadataStore(redisConnectionFactory)
    }

    // 配置持久化文件过滤器
    @Bean
    fun sftpPersistentFilter(metadataStore: MetadataStore): FileListFilter<ChannelSftp.LsEntry> {
        val filter = SftpPersistentAcceptOnceFileListFilter(metadataStore, "fileKeyPrefix")
        filter.forRecursion = true // 启用递归处理
        return filter
    }

    FTP入站通道适配器
    @Bean
    fun sftpInboundAdapter(sessionFactory: SessionFactory<LsEntry>,
                           filter: FileListFilter<LsEntry>): SftpInboundFileSynchronizingMessageSource {
        return Sftp.inboundAdapter(sessionFactory)
            .remoteDirectory("/remote/files")
            .filter(filter) // 注入过滤器
            .get()
    }
}

3. 递归处理配置注意事项

// 正确配置递归过滤器
val filter = FtpPersistentAcceptOnceFileListFilter(metadataStore, "ftpFiles")
filter.apply {
    forRecursion = true  // 启用递归处理
    alwaysAcceptDirectories = true // 自动设置
}

// 错误配置示例
val badFilter = FtpPersistentAcceptOnceFileListFilter(metadataStore, "ftpFiles")
badFilter.forRecursion = false // 深层目录变更无法检测

启用递归的关键影响

设置 forRecursion=true 会导致：

1. 使用完整文件路径作为元数据键
2. 现有存储中非顶级目录的键将失效
3. 目录遍历行为改变（始终全量扫描）

最佳实践与故障处理

1. 灾难性故障恢复方案

@Service
class FileRecoveryService(
    private val metadataStore: MetadataStore
) {
    fun recoverFailedFile(filePath: String) {
        // 从元数据存储中手动移除故障文件
        metadataStore.remove("fileKeyPrefix/$filePath")
        logger.info("已恢复文件: $filePath")
    }
}

CAUTION

在电源故障或服务崩溃后，当前处理文件可能残留在过滤器中，需要手动清除！

2. 复合过滤器配置规范

// 正确配置复合过滤器 (所有委托过滤器必须支持单文件过滤)
@Bean
fun compositeFilter(): CompositeFileListFilter<LsEntry> {
    val filter1 = SftpPersistentAcceptOnceFileListFilter(metadataStore, "prefix1")
    val filter2 = SftpRegexPatternFileListFilter(Pattern.compile(".*\\.txt$"))

    // 验证所有过滤器支持单文件操作
    check(filter1.supportsSingleFileFiltering() && filter2.supportsSingleFileFiltering())

    return CompositeFileListFilter(listOf(filter1, filter2))
}

// 错误配置示例
@Bean
fun badCompositeFilter(): CompositeFileListFilter<LsEntry> {
    val persistentFilter = SftpPersistentAcceptOnceFileListFilter(metadataStore, "prefix")
    val markerFilter = SftpSimplePatternFileListFilter("*.MARKER")

    // markerFilter不支持单文件过滤将导致回退旧行为
    return CompositeFileListFilter(listOf(persistentFilter, markerFilter))
}

实际应用场景

分布式订单处理系统

生产环境建议

1. 对元数据存储键使用业务相关前缀（如 region/order_)
2. 设置定期清理任务移除过期条目
3. 监控元数据存储大小，避免无限增长

常见问题解决方案

Q1: 文件被跳过但实际未处理

原因：元数据存储残留记录
解决：检查存储中对应键值并手动删除

// 诊断工具方法
fun checkFileStatus(filePath: String) {
    val key = "fileKeyPrefix/$filePath"
    val value = metadataStore.get(key)
    println("文件状态: ${value ?: "未记录"}")
}

Q2: 目录更新未检测到

原因：forRecursion=false 且目录结构深
解决：启用递归处理模式

// 修正配置
filter.forRecursion = true

Q3: 集群节点处理相同文件

原因：元数据存储未正确共享
解决：验证存储实现（如 Redis）的连接配置

@Bean
fun redisConnectionFactory(): RedisConnectionFactory {
    return LettuceConnectionFactory().apply {
        hostName = "redis-cluster.example.com"
        port = 6379
        password = "securePassword"
    }
}

总结

远程持久文件列表过滤器是 Spring Integration 文件处理的核心安全机制。关键要点：

1. ✅ 使用共享 MetadataStore 实现集群安全
2. ⚠️ 灾难性故障后需手动干预
3. 🔄 启用 forRecursion=true 解决深层目录问题
4. ⚡ 复合过滤器需全部组件支持单文件过滤

NOTE

随着 Spring Integration 的版本更新，持久过滤器的默认行为可能改变，建议定期查阅官方文档获取最新信息。