Spring Integration文件轮询如何重复处理已处理过的文件？

针对你遇到的文件轮询无法重复处理已处理文件的问题，以下是可行的解决思路和具体实现方案：

核心原因 #

Spring Integration 文件轮询默认使用 AcceptOnceFileListFilter，该过滤器会维护已处理文件的状态缓存，默认不会重复选取已处理过的文件。你的需求需要替换或修改这个默认过滤器逻辑。

方案1：使用内置的 AlwaysAcceptFileListFilter（最简洁） #

Spring Integration 提供了直接接受所有文件的过滤器，无需自定义，直接替换默认过滤器即可实现重复处理：

@Bean
public IntegrationFlow filePollingFlow() {
    return IntegrationFlows.from(
            Files.inboundAdapter(new File("/你的目标目录路径"))
                    // 替换默认过滤器，允许所有文件被重复处理
                    .filter(new AlwaysAcceptFileListFilter()),
            e -> e.poller(Pollers.fixedDelay(5000)) // 轮询间隔，按需调整
        )
        .handle((payload, headers) -> {
            // 你的文件处理逻辑
            File file = (File) payload;
            // 读取文件、传递元数据等操作
            return null;
        })
        .get();
}

方案2：自定义 FileListFilter（灵活控制重复逻辑） #

如果需要根据业务规则（比如文件修改时间、内容变更等）决定是否重复处理，可实现自定义过滤器。注意必须通过 .filter() 方法替换默认过滤器，否则自定义逻辑不会生效：

自定义过滤器实现 #

public class CustomRepeatFileFilter implements FileListFilter<File> {
    @Override
    public List<File> filterFiles(File[] files) {
        // 示例逻辑：仅处理最近1小时内修改过的文件（支持重复处理更新后的文件）
        long oneHourAgo = System.currentTimeMillis() - 3600000;
        return Arrays.stream(files)
                .filter(file -> file.lastModified() > oneHourAgo)
                .collect(Collectors.toList());
        
        // 如果需要无条件重复处理所有文件，直接返回：
        // return Arrays.asList(files);
    }
}

配置到轮询流程 #

@Bean
public IntegrationFlow filePollingFlow() {
    return IntegrationFlows.from(
            Files.inboundAdapter(new File("/你的目标目录路径"))
                    // 替换为自定义过滤器
                    .filter(new CustomRepeatFileFilter()),
            e -> e.poller(Pollers.fixedDelay(5000))
        )
        .handle(...) // 你的处理逻辑
        .get();
}

方案3：重置默认过滤器状态（按需重复处理） #

如果只是需要定期重置已处理文件的记录，可保留 AcceptOnceFileListFilter 并定时清空其状态：

@Bean
public AcceptOnceFileListFilter<File> fileListFilter() {
    return new AcceptOnceFileListFilter<>();
}

@Bean
public IntegrationFlow filePollingFlow() {
    return IntegrationFlows.from(
            Files.inboundAdapter(new File("/你的目标目录路径"))
                    .filter(fileListFilter()),
            e -> e.poller(Pollers.fixedDelay(5000))
        )
        .handle(...) // 你的处理逻辑
        .get();
}

// 定时重置过滤器缓存，示例：每天凌晨执行
@Scheduled(cron = "0 0 0 * * ?")
public void resetProcessedFiles() {
    fileListFilter().clear();
}

关于 preventDuplicates=false 未生效的排查 #

如果设置该参数后无效果，大概率是因为：

1. 代码中同时配置了自定义过滤器，覆盖了该参数的默认行为；
2. 配置位置错误，需确保在 Files.inboundAdapter() 链式调用中设置，而非后续流程。

正确配置示例：

Files.inboundAdapter(new File("/你的目标目录路径"))
     .preventDuplicates(false)

自定义过滤器未生效的排查步骤 #

1. 确认使用 .filter() 方法替换默认过滤器，而非添加额外过滤器；
2. 开启 Spring Integration DEBUG 日志，查看过滤器的调用日志，确认逻辑是否执行；
3. 检查是否存在其他流程组件（如后续的消息过滤器）拦截了重复处理的文件。

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