如何实现无线程重叠的Scheduled MultiTasking多线程文件夹定时任务

问题描述 #

我现有5个参数化线程分别对应A、B、C、D、E共5个文件夹，线程入参为对应文件夹名称，每个文件夹下包含多个需执行相同操作的文件，操作顺序为对每个文件依次执行Read、Process、Write，整套流程需要定时持续运行。
此前我尝试过两种方案均存在问题：

1. 搭配MultiThreading实现定时调度：5个参数化线程执行同一个Task，会出现线程重叠问题，无法得到预期输出。
2. 搭配MultiTasking实现定时调度：为每个文件夹单独创建实现Runnable接口的类，通过executor.scheduleAtFixedRate调度每个类实例；若将newScheduledThreadPool的corePoolSize设为1，会出现同步阻塞问题，第一个文件夹的操作未完成时其余4个文件夹无法启动处理；若增大corePoolSize则会出现和第一种方案相同的线程重叠问题。

现寻求可行的解决方案，实现底层任务无线程重叠的调度效果。

方案2的示例代码如下： #

class FolderA implements Runnable{
    private final String fileName;

    FolderA(String fileName){
        this.fileName=fileName;
    }
    @Override
    public void run() {
        ScheduleJob.insideRun(fileName);
    }
}

class FolderB implements Runnable{...}
class FolderC implements Runnable{...}
class FolderD implements Runnable{...}
class FolderE implements Runnable{...}

public class ScheduleJob{
    ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
    
    FolderA folderA = new FolderA("folderA");
    executor.scheduleAtFixedRate(folderA, 60,60, TimeUnit.SECONDS);
    
    FolderB folderB = new FolderB("folderB");
    executor.scheduleAtFixedRate(folderB, 60,60, TimeUnit.SECONDS);
    
    // 剩余C、D、E文件夹同理初始化
    
     public static void insideRun(String folderName){
     
     // 包含对当前文件夹下所有文件执行的各类操作
     // 执行顺序为：针对每个文件依次执行Read、Process、Write
     
     }
    
}

可行解决方案 #

核心思路 #

你之前遇到的问题根源是scheduleAtFixedRate的原生特性：若任务执行时长超过设定的调度间隔，该任务的下一次执行会在上次结束后立即触发，只要池内有空闲线程就会出现同个文件夹的任务并发执行；而单核心池会强制所有文件夹任务串行，完全浪费并行能力。
我们可以通过给每个文件夹任务加独立运行状态标记的方式解决，具体逻辑如下：

• 调度池核心线程数设为5，支持5个文件夹任务并行执行，互相不阻塞
• 每个任务内部维护一个volatile的运行状态标记，触发调度时先检查状态：如果当前任务正在运行，直接跳过本次调度；如果未运行，先标记为运行中，再执行文件处理逻辑，执行完成后改回未运行状态
• 复用同一套Runnable实现，无需为每个文件夹单独写实现类，减少冗余代码

实现代码 #

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

// 可复用的文件夹处理任务类
class FolderProcessTask implements Runnable {
    private final String folderName;
    // 标记当前文件夹任务是否正在运行，volatile保证多线程下状态可见性
    private volatile boolean isRunning = false;

    public FolderProcessTask(String folderName) {
        this.folderName = folderName;
    }

    @Override
    public void run() {
        // 任务已在运行，直接跳过本次调度
        if (isRunning) {
            return;
        }
        try {
            isRunning = true;
            // 执行原有文件处理逻辑
            ScheduleJob.insideRun(folderName);
        } finally {
            // 无论处理是否抛出异常，最终都要重置状态
            isRunning = false;
        }
    }
}

public class ScheduleJob {
    // 核心线程数设为5，刚好支持5个文件夹任务同时运行
    private static final ScheduledExecutorService executor = Executors.newScheduledThreadPool(5);

    public static void main(String[] args) {
        // 初始化5个文件夹对应的任务实例
        String[] folders = {"folderA", "folderB", "folderC", "folderD", "folderE"};
        for (String folder : folders) {
            FolderProcessTask task = new FolderProcessTask(folder);
            // 初始延迟60秒，之后每60秒调度一次
            executor.scheduleAtFixedRate(task, 60, 60, TimeUnit.SECONDS);
        }
    }

    public static void insideRun(String folderName) {
        // 原有针对文件夹下文件的Read、Process、Write逻辑保持不变
    }
}

效果说明 #

• 不同文件夹的任务可以并行执行，不会互相阻塞，资源利用率更高
• 同一个文件夹的任务不会重叠执行：如果某次处理时长超过60秒，下一次调度触发时发现任务正在运行就会跳过，等到下一个60秒周期再检查执行
• 代码冗余度低，后续新增文件夹只需在folders数组中新增对应名称即可，无需新增类

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