Apache Camel-Spring Boot 基于时间的文件可用性检测求助：路由冻结异常问题

我完全理解你现在遇到的困扰——加入定时文件可用性检测逻辑后，原本实时的文件同步路由会在定时任务触发时突然“冻结”，期间新拷贝到源目录的文件完全不处理，等缓冲时间结束后才批量执行，这严重影响了同步的稳定性。我们来一步步拆解问题根源，再给出针对性的解决办法。

问题根源分析 #

从你的代码和描述来看，核心问题出在同步阻塞的延迟逻辑和潜在的线程安全风险上：

1. 你用了seda + delayer组合实现缓冲等待，但Camel的delayer默认是同步阻塞的——seda路由的消费者线程会被直接挂起，直到缓冲时间结束。如果你的缓存服务（localCacheService）线程安全不足，或者主文件路由与定时路由共享了关键线程资源，就会导致主文件消费线程被意外阻塞，出现“冻结”现象。
2. 主文件路由的file组件默认是单线程轮询，若此时线程因资源竞争被占用，自然无法处理新文件，直到阻塞线程释放资源。

针对性解决方案 #

我们可以通过异步延迟逻辑、线程安全加固、线程池隔离三个层面来修复问题，这里提供两种可行的实现方案：

方案1：改造seda + delayer为异步模式（快速适配现有代码）

保留你现有seda的结构，但启用delayer的异步模式，同时配置seda的并发线程，避免线程阻塞影响主路由：

private void invokeBufferedCheck(RouteBuilder routeBuilder, String fileName, int bufferMinutes, long fileExpectedTime) {
    // 给seda配置并发消费者，避免单线程阻塞
    String sedaEndpoint = "seda:bufferDelay-" + fileName + "?concurrentConsumers=2";
    routeBuilder.from(sedaEndpoint)
        .routeId("BufferedCheckRoute-" + fileName + "-" + UUID.randomUUID())
        .log("Buffer check started for file: " + fileName + ", waiting for {} minutes", bufferMinutes)
        // 启用异步延迟：不会阻塞当前线程，后台用独立线程处理等待逻辑
        .delayer(bufferMinutes * 60 * 1000L, true)
        .log("Buffer delay expired for file: " + fileName)
        .process(exchange -> {
            Long fileProcessedTime = localCacheService.get(fileName);
            boolean stillMissing = (fileProcessedTime == null) || (fileProcessedTime < fileExpectedTime);
            exchange.setProperty("stillMissing", stillMissing);
            exchange.setProperty("fileName", fileName);
        })
        .choice()
            .when(routeBuilder.simple("${exchangeProperty.stillMissing} == true"))
                .log("ALERT: File ${exchangeProperty.fileName} still missing after buffer wait.")
                // 在这里添加发送告警邮件的逻辑
            .otherwise()
                .log("File ${exchangeProperty.fileName} arrived within buffer window.")
                .process(e -> localCacheService.remove(fileName))
        .end();
}

同时，务必确保localCacheService是线程安全的，比如用ConcurrentHashMap作为底层存储：

@Service
public class LocalCacheService {
    private final ConcurrentHashMap<String, Long> fileProcessedCache = new ConcurrentHashMap<>();

    public boolean containsKey(String fileName) {
        return fileProcessedCache.containsKey(fileName);
    }

    public Long get(String fileName) {
        return fileProcessedCache.get(fileName);
    }

    public void put(String fileName, Long processedTime) {
        fileProcessedCache.put(fileName, processedTime);
    }

    public void remove(String fileName) {
        fileProcessedCache.remove(fileName);
    }
}

方案2：用Quartz2实现非阻塞的一次性延迟任务（更稳定）

如果想彻底避免线程阻塞问题，推荐用Quartz2的一次性定时任务替代seda + delayer，完全隔离定时逻辑与主文件路由的线程资源：

1. 先新增全局缓冲检查路由：

private void configureGlobalBufferedCheckRoute(RouteBuilder routeBuilder) {
    routeBuilder.from("quartz2://fileCheckGroup?job.name=BufferCheckJob*")
        .routeId("GlobalBufferedCheckRoute")
        .log("Starting buffer check for file: ${header.fileName}")
        .process(exchange -> {
            String fileName = exchange.getIn().getHeader("fileName", String.class);
            long fileExpectedTime = exchange.getIn().getHeader("fileExpectedTime", Long.class);
            Long fileProcessedTime = localCacheService.get(fileName);
            boolean stillMissing = (fileProcessedTime == null) || (fileProcessedTime < fileExpectedTime);
            exchange.setProperty("stillMissing", stillMissing);
            exchange.setProperty("fileName", fileName);
        })
        .choice()
            .when(routeBuilder.simple("${exchangeProperty.stillMissing} == true"))
                .log("ALERT: File ${exchangeProperty.fileName} still missing after buffer window.")
                // 在这里添加发送告警邮件的逻辑
            .otherwise()
                .log("File ${exchangeProperty.fileName} arrived within buffer window.")
                .process(e -> localCacheService.remove(e.getProperty("fileName", String.class)))
        .end();
}

2. 修改原有的invokeFileCheck方法，替换seda发送为Quartz2一次性任务调度：

private void invokeFileCheck(RouteBuilder routeBuilder, String fileName, String checkTime, long fileExpectedTime, String quartzId) {
    routeBuilder.from(quartzId)
        .routeId("FileCheckRoute-" + fileName + "-" + UUID.randomUUID())
        .log("Checking for file: " + fileName + " at scheduled time: " + checkTime)
        .process(exchange -> {
            boolean containsKey = localCacheService.containsKey(fileName);
            if (containsKey) {
                Long fileProcessedTime = localCacheService.get(fileName);
                if (fileProcessedTime >= fileExpectedTime) {
                    exchange.setProperty("fileName", fileName);
                    exchange.setProperty("fileExists", true);
                }
            } else {
                exchange.setProperty("fileName", fileName);
                exchange.setProperty("fileExists", false);
            }
        })
        .choice()
            .when(routeBuilder.simple("${exchangeProperty.fileExists} == false"))
                .log("File ${exchangeProperty.fileName} is missing. Scheduling buffer check after {} minutes", bufferMinutes)
                .process(exchange -> {
                    // 生成唯一任务ID，避免冲突
                    String jobName = "BufferCheckJob-" + fileName + "-" + UUID.randomUUID();
                    // 计算缓冲后的触发时间
                    long triggerTime = System.currentTimeMillis() + (bufferMinutes * 60 * 1000L);
                    // 构建Quartz2一次性任务端点
                    String quartzEndpoint = String.format(
                        "quartz2://fileCheckGroup/%s?trigger.repeatCount=0&trigger.startTime=%d",
                        jobName, triggerTime
                    );
                    // 传递必要参数
                    exchange.getIn().setHeader("fileName", fileName);
                    exchange.getIn().setHeader("fileExpectedTime", fileExpectedTime);
                    // 异步发送任务
                    exchange.getContext().createProducerTemplate().send(quartzEndpoint, exchange);
                })
            .otherwise()
                .log("File ${exchangeProperty.fileName} found. No need to wait.")
                .process(e -> localCacheService.remove(fileName))
        .end();
}

3. 在configureRoute中先初始化全局缓冲路由：

@Override
public void configureRoute(RouteBuilder routeBuilder) throws Exception {
    // ... 原有主文件同步路由代码 ...

    if (localtolocalProps.getFilechecks() == null) {
        return;
    }
    // 先初始化全局缓冲检查路由
    configureGlobalBufferedCheckRoute(routeBuilder);

    for (LocalToLocal.FileCheck fileCheck : localtolocalProps.getFilechecks()) {
        // ... 原有定时检查路由初始化代码 ...
    }
}

额外优化建议 #

给主文件路由的file组件配置多线程消费，进一步提升稳定性：
在你的fromCamelOptions中添加consumer.threads=5，让文件消费使用多线程，即使个别线程出现意外，其他线程仍能继续处理文件：

# 示例配置
localtolocal:
  source: /path/to/source
  destination: /path/to/dest
  fromCamelOptions:
    consumer.threads: 5
    consumer.delay: 1000 # 每秒轮询一次源目录

内容来源于stack exchange