java - 自定义多线程 : limiting number of tasks of some type to be executed in parallel, 不限制其他类型任务

Question

我有 3 种类型的任务:A、B、C。

我想在 N 个线程中并行运行这些任务。假设任务列表如下:

A, B, C, B, C, A, B, C

当然，我可以使用ExecutorService实现多线程执行，但问题是我一次最多需要执行一个C类型的任务 .C 类型的其他任务必须按顺序执行，但与任务 A 和/或 B 并行执行。

例如，3线程执行器可能处于以下任何状态:

A B C
A A A
A C B
B B C
B B B
B C
A C 
A B 
C  
... 

(允许同时执行多个A或B类型的任务，但一次最多执行一个C类型的任务)

有什么方法可以在Java中实现这一点吗？

更新

这是我想到的，这是一种正确的方法吗？ 在这里，我通过 ExecutorService 执行所有任务。在执行时，我将检查是否有任何其他 C 任务正在运行。如果没有，我将执行，否则我会将其添加到将被出队的队列中成功完成任何其他任务，并且我还检查任何 C 任务是否正在运行

public class Test {

public void startExecution() {
    Queue<String> runQ = new LinkedList<>();
    ThreadPool exec = (ThreadPool) Executors.newFixedThreadPool(RunSettings.getRunSettings().getThreadCount());
    while (!runQ.isEmpty() && !SystemDefaults.stopExecution.get()) {
        String TaskName = runQ.remove();
        Task t = new Task(TaskName);
        exec.execute(t, TaskName);

    }
    exec.shutdown();
    if (exec.awaitTermination(RunSettings.getRunSettings().getExecutionTimeOut(), TimeUnit.MINUTES)) {
        System.out.println("[CONTROL: ALL TEST TASKS COMPLETED SUCCESSFULLY.]");
    } else {
        System.out.println("[CONTROL: ALL THE TEST TASKS DID NOT COMPLETE SUCCESSFULLY IN STIPULATED TIME. FORCEFULLY FINALIZING.]");
        exec.shutdownNow();

    }
}
}

我创建的线程池

public class ThreadPool extends ThreadPoolExecutor {

public ThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
}

final String CTask = "TaskC";
Map<Runnable, String> TaskPool = new HashMap<>();
Queue<Runnable> TaskCList = new LinkedList<>();

@Override
protected void afterExecute(Runnable r, Throwable t) {
    super.afterExecute(r, t);
    if (TaskPool.containsKey(r)) {
        TaskPool.remove(r);
    }
    if (!TaskPool.containsValue(CTask) && !TaskCList.isEmpty()) {
        Runnable ieRun = TaskCList.remove();
        super.execute(ieRun);
        TaskPool.put(ieRun, CTask);
    }
}

public void execute(Runnable command, String TaskType) {
    if (TaskPool.containsValue(TaskType)
            && TaskType.equalsIgnoreCase(CTask)) {
        System.out.println("Another Instance of  " + CTask + " Running");
        TaskCList.add(command);
    } else {
        super.execute(command);
        TaskPool.put(command, TaskType);
    }
}

}

Answer 1

最简单的方法是创建 2 个执行器:一个单线程用于 C 类型的任务，另一个多线程用于其他类型的任务:

class ExecutorWrapper {
    private ExecutorService forC = Executors.newSingleThreadExecutor();
    private ExecutorService forAnother = Executors.newFixedThreadPool(THREAD_NUMBER);

    public void acceptTask(Runnable r) {
        if (r instanceof TaskC) {
            forC.execute(r);
        } else {
            forAnother.execute(r);
        }
    }
}

现在，任何 C 类型的任务都将在 forC 执行器内部队列中等待，直到另一个此类类型的任务完成。

如果您不想创建另一个执行程序，则需要实现某种并发控制，这要复杂得多，并且由于可能发生竞争条件而难以调试。我可以提出解决方案草案，但没有代码:

1. 创建一个标志来指示另一个任务 C 是否已在执行，以及另一个任务 C 将等待的队列
2. 当类型C的任务到达时，检查是否有另一个任务C正在执行，如果是，则将其添加到指定队列中
3. 任务 C 完成后，发送某种任务 C 已完成的通知 - 以便从上述队列中获取下一个任务 C 并将其发送到执行器中。如果队列为空，则清除该标志以指示现在没有任务 C 正在执行。此类通知可以通过使用 Callable 包装任务 C 并调用将阻塞直到任务完成的 Future#get 方法来实现。