Java EE桌面应用中带复杂约束的多源数据文件线程池处理架构咨询

这个场景确实很有代表性——既要靠多线程榨取性能，又得严格保证同一数据源的文件串行、按序处理，还得应对数据源异步生成文件的情况。我给你分享几个基于JDK原生API的可行方案，都能满足你的约束：

方案一：按数据源分组的队列+全局线程池+数据源锁 #

这是最直观也最易维护的方案，核心思路是给每个数据源单独维护一个有序任务队列，用全局线程池处理任务，同时通过锁保证同一数据源的任务串行执行：

1. 核心组件：

   • ConcurrentHashMap<String, BlockingQueue<FileTask>> dataSourceQueues：键是数据源唯一ID，值是该数据源的文件任务队列，用PriorityBlockingQueue来自动按文件创建时间排序，确保任务顺序。
   • ConcurrentHashMap<String, ReentrantLock> dataSourceLocks：每个数据源对应一把锁，保证同一时间只有一个线程能处理该数据源的任务。
   • ExecutorService workerPool = Executors.newFixedThreadPool(15)：你的15线程工作池。
   • 一个文件扫描线程/定时任务：定期扫描目录，解析文件名提取数据源ID，将文件封装成FileTask（包含数据源ID、文件、创建时间），放入对应数据源的PriorityBlockingQueue。

2. 任务执行逻辑：
   工作线程循环从所有数据源队列中尝试获取任务（或者用一个调度线程把队列中的任务提交到线程池），拿到任务后：

   String dataSourceId = task.getDataSourceId();
   ReentrantLock lock = dataSourceLocks.computeIfAbsent(dataSourceId, k -> new ReentrantLock());
   // 获取锁，保证同一数据源串行
   lock.lock();
   try {
       // 处理文件：解析、业务逻辑、删除文件
       processFile(task.getFile());
       // 标记该文件已处理（可选，避免重复扫描）
       markFileAsProcessed(dataSourceId, task.getFile().getName());
   } finally {
       lock.unlock();
   }
   

3. 优势：

   • 严格保证同一数据源的文件按时间顺序处理，因为PriorityBlockingQueue会自动排序。
   • 线程池资源利用率高，不同数据源的任务可以并行处理。
   • 实现简单，依赖JDK原生类，不需要第三方库。

方案二：全局优先级队列+数据源锁 #

如果不想维护大量的数据源队列，可以用一个全局的优先级队列，结合数据源锁来控制串行：

1. 核心组件：

   • PriorityBlockingQueue<FileTask> globalQueue：全局任务队列，FileTask实现Comparable，先按数据源ID分组，再按创建时间排序，确保同一数据源的任务按顺序排列。
   • ConcurrentHashMap<String, ReentrantLock> dataSourceLocks：和方案一一样的数据源锁。
   • ExecutorService workerPool = Executors.newFixedThreadPool(15)：工作线程池。

2. 任务执行逻辑：
   工作线程从全局队列取任务，尝试获取对应数据源的锁：

   while (!Thread.currentThread().isInterrupted()) {
       FileTask task = globalQueue.take();
       String dataSourceId = task.getDataSourceId();
       ReentrantLock lock = dataSourceLocks.computeIfAbsent(dataSourceId, k -> new ReentrantLock());
       
       // 尝试获取锁，获取不到就把任务放回队列，避免阻塞
       if (lock.tryLock()) {
           try {
               processFile(task.getFile());
               markFileAsProcessed(dataSourceId, task.getFile().getName());
           } finally {
               lock.unlock();
           }
       } else {
           // 放回队列，等待下次尝试
           globalQueue.put(task);
           // 可选：短暂休眠，避免频繁抢锁
           Thread.sleep(100);
       }
   }
   

3. 注意点：

   • 任务可能会被多次取出放回，在高并发场景下可能有轻微的性能损耗，但对于你的文件处理场景（单文件3-15秒）来说，这个损耗可以忽略。
   • 全局队列的排序要正确，确保同一数据源的任务按时间顺序排列。

方案三：基于单线程执行器的分组调度 #

这个方案利用每个数据源对应一个单线程执行器，然后把这些单线程执行器的任务提交到全局线程池，本质是把串行逻辑下放到每个数据源的执行器：

1. 核心组件：

   • ConcurrentHashMap<String, Executor> dataSourceExecutors：每个数据源对应一个Executors.newSingleThreadExecutor()，保证该数据源的任务串行执行。
   • ExecutorService globalPool = Executors.newFixedThreadPool(15)：全局线程池，用来承载所有单线程执行器的任务。
   • 文件扫描线程：和之前一样，发现新文件后，提交到对应数据源的单线程执行器。

2. 任务提交逻辑：

   String dataSourceId = extractDataSourceId(fileName);
   Executor executor = dataSourceExecutors.computeIfAbsent(dataSourceId, k -> 
       Executors.newSingleThreadExecutor(new ThreadPoolExecutor.CallerRunsPolicy())
   );
   executor.execute(() -> {
       // 这里的任务会被串行执行，不需要额外锁
       processFile(file);
       markFileAsProcessed(dataSourceId, file.getName());
   });
   

3. 优势：

   • 不需要手动管理锁，单线程执行器天然保证串行。
   • 代码更简洁，任务提交逻辑清晰。
   • 可以给每个数据源的执行器设置单独的超时、重试策略，灵活性更高。

关键补充点 #

• 文件去重：因为无法和数据源同步，扫描线程可能会重复扫描到已处理的文件，所以需要一个全局的已处理文件记录（比如ConcurrentHashMap<String, Set<String>> processedFiles，键是数据源ID，值是已处理的文件名），扫描时先判断是否已处理。
• 异常处理：文件处理失败时，建议把失败任务放到单独的重试队列，避免阻塞正常任务，重试次数可以配置。
• 队列监控：可以给每个队列添加监控，比如队列长度、任务处理耗时，方便排查性能问题。

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