Spring Integration：避免文件重复拾取 处理完支持读取同名文件

当前配置存在三个核心问题，才会导致重复拾取、同名新文件无法识别的现象：

• 轮询配置不合理：fixed-delay="5"单位为毫秒，轮询频率远高于200MB文件10-20秒的处理耗时，文件还在处理阶段就会被多次扫描
• 去重逻辑缺陷：prevent-duplicates="true"时默认使用AcceptOnceFileListFilter，仅在内存中记录已处理的文件名，后续投放的同名文件会被直接过滤；手动设为false后又缺少文件状态校验，未处理完成的文件会被重复投递
• 缺少串行处理控制：没有对轮询线程、通道做单线程约束，无法保证当前文件处理完成后再拉取下一个文件，也没有文件锁机制避免并发读取同一文件

按照以下步骤调整配置即可满足要求，保证单文件顺序处理、不重复拾取、同名新文件可正常识别：

1. 替换组合文件过滤器，替代原生的prevent-duplicates逻辑
   不要依赖默认的prevent-duplicates开关，改用三层校验逻辑：
   • 新增LastModifiedFileListFilter，仅拾取最后修改时间超过3秒的文件，避免文件还在写入传输过程中就被扫描到
   • 新增FileSystemPersistentAcceptOnceFileListFilter，基于「文件路径+最后修改时间+文件大小」的元数据判断是否已处理，而非仅记录文件名，可正常识别内容更新的同名新文件
   • 新增NioFileLocker文件锁，给正在处理的文件加系统级锁，处理完成前不会被重复拾取
2. 调整轮询模型，保证单线程拉取
   把轮询间隔调整为1000毫秒（1秒），指定容量为1的独立线程池执行轮询任务，避免多线程并发拉取文件。fixed-delay本身的逻辑是上一次轮询任务执行完成后才开始计算等待间隔，配合单线程池可避免轮询任务堆叠。
3. 使用直连通道绑定消费端
   显式声明文件通道为直连通道，让文件拉取和业务处理在同一个线程执行，只有handleFlow方法执行完成（当前文件处理完毕），才会触发下一次轮询拉取，严格实现串行处理。

<bean id="test-file-bean" class="com.test.MyHandler"/>

<!-- NIO文件锁，处理中文件自动加锁，避免重复读取 -->
<bean id="fileLocker" class="org.springframework.integration.file.locking.NioFileLocker"/>

<!-- 组合文件过滤器 -->
<bean id="test-file-filter" class="org.springframework.integration.file.filters.CompositeFileListFilter">
    <constructor-arg>
        <list>
            <!-- 过滤写入中文件：仅拾取落盘超过3秒的文件，可根据实际传输速度调整age值 -->
            <bean class="org.springframework.integration.file.filters.LastModifiedFileListFilter">
                <property name="age" value="3000"/>
            </bean>
            <!-- 元数据去重：基于文件属性判断是否已处理，支持同名新文件识别 -->
            <bean class="org.springframework.integration.file.filters.FileSystemPersistentAcceptOnceFileListFilter">
                <constructor-arg>
                    <bean class="org.springframework.integration.metadata.SimpleMetadataStore"/>
                </constructor-arg>
                <constructor-arg value="test-file-handle-"/>
            </bean>
        </list>
    </constructor-arg>
</bean>

<!-- 单线程轮询池，保证同一时间只有一个轮询任务在执行 -->
<task:executor id="filePollExecutor" pool-size="1"/>

<int-file:inbound-channel-adapter
        id="test-adapter-inbound"
        directory="${my.test.dir}"
        channel="test-file-channel"
        filter="test-file-filter"
        locker="fileLocker"
        prevent-duplicates="false"
        auto-startup="true"
        auto-create-directory="true">
    <!-- 上次任务完成后等待1秒再执行下一次轮询，单线程执行避免并发 -->
    <int:poller fixed-delay="1000" time-unit="MILLISECONDS" task-executor="filePollExecutor"/>
</int-file:inbound-channel-adapter>

<!-- 直连通道，拉取与处理同线程，严格串行执行 -->
<int:channel id="test-file-channel"/>

<int:service-activator
        input-channel="test-file-channel"
        ref="test-file-bean"
        method="handleFlow"/>

• 如果需要应用重启后不重复处理已扫描过的文件，可将过滤器中使用的SimpleMetadataStore替换为PropertiesPersistingMetadataStore，将文件处理记录持久化到本地磁盘
• 原配置中5ms的轮询间隔无实际意义，过高的扫描频率只会带来不必要的磁盘IO消耗，1秒的间隔完全满足常规文件投递的实时性要求
• 该配置下MyHandler的handleFlow方法执行完成前，不会拉取新的文件，完全符合单文件顺序处理的要求

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