如何为SFTP检测到的每个文件独立设置preMove延迟？

你当前配置里的delay=30是SFTP组件的目录轮询间隔，意思是每隔30秒去扫描一次目录，并不是给每个文件单独设置上传后的延迟，所以达不到你要的按文件上传时间分别延迟30秒的效果。下面给出两种可行的实现方案（以Apache Camel为例，这是SFTP路由场景最常用的框架）：

方案一：基于文件修改时间计算延迟（同步处理） #

通过获取文件的最后修改时间，计算当前时间与修改时间的差值，不足30秒则补足延迟后再处理文件。

Java DSL路由示例 #

from("sftp://username@host:port/path/to/import/directory?password=password&moveFailed=/path/to/error/directory&move=/path/to/success/directory&preMove=/path/to/inProgress/directory&readLock=changed")
    .process(exchange -> {
        File targetFile = exchange.getIn().getBody(File.class);
        long uploadFinishTime = targetFile.lastModified();
        long elapsed = System.currentTimeMillis() - uploadFinishTime;
        long needDelay = 30000 - elapsed;
        if (needDelay > 0) {
            Thread.sleep(needDelay);
        }
    })
    // 这里添加你的后续处理逻辑
    .to("your-process-endpoint");

XML配置示例 #

先定义处理器Bean：

public class FileDelayProcessor implements Processor {
    @Override
    public void process(Exchange exchange) throws Exception {
        File targetFile = exchange.getIn().getBody(File.class);
        long uploadFinishTime = targetFile.lastModified();
        long elapsed = System.currentTimeMillis() - uploadFinishTime;
        long needDelay = 30000 - elapsed;
        if (needDelay > 0) {
            Thread.sleep(needDelay);
        }
    }
}

再配置路由：

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:camel="http://camel.apache.org/schema/spring"
       xsi:schemaLocation="
           http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
           http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">

    <bean id="fileDelayProcessor" class="com.yourpackage.FileDelayProcessor"/>

    <camelContext xmlns="http://camel.apache.org/schema/spring">
        <route>
            <from uri="sftp://username@host:port/path/to/import/directory?password=password&moveFailed=/path/to/error/directory&move=/path/to/success/directory&preMove=/path/to/inProgress/directory&readLock=changed"/>
            <process ref="fileDelayProcessor"/>
            <!-- 后续处理节点 -->
            <to uri="your-process-endpoint"/>
        </route>
    </camelContext>
</beans>

方案二：使用SEDA队列异步延迟（推荐） #

这种方式不会阻塞SFTP的轮询线程，把文件消息发送到异步队列，每个消息根据上传时间设置单独延迟，更适合高并发场景。

Java DSL路由示例 #

// SFTP扫描路由
from("sftp://username@host:port/path/to/import/directory?password=password&moveFailed=/path/to/error/directory&move=/path/to/success/directory&preMove=/path/to/inProgress/directory&readLock=changed")
    .process(exchange -> {
        File targetFile = exchange.getIn().getBody(File.class);
        long uploadFinishTime = targetFile.lastModified();
        long elapsed = System.currentTimeMillis() - uploadFinishTime;
        long delayMillis = Math.max(0, 30000 - elapsed);
        // 设置SEDA队列的延迟属性
        exchange.setProperty(Exchange.SEDA_DELAY, delayMillis);
    })
    .to("seda:delayedFileProcessing");

// 延迟后处理路由
from("seda:delayedFileProcessing")
    // 后续处理逻辑
    .to("your-process-endpoint");

关键注意事项 #

• 必须配置readLock=changed（或其他合适的读锁策略），确保文件完全上传完成后才被读取，避免处理不完整的文件。
• 方案一的同步延迟会占用轮询线程，如果文件较多可能导致扫描延迟，因此优先推荐方案二。
• 可以移除原配置中的delay=30&initialDelay=30&useFixedDelay=true参数，根据需要调整SFTP的轮询间隔（比如设置delay=5让扫描更频繁）。

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