java - 使用共享的 ExecutorService 作为任务队列，我如何知道作业何时完成？

Question

我有一个 JobService 可以处理更大的作业。作业会动态地分割为多个任务，任务还可能生成子任务等，因此无法预测作业的任务总数。每个任务都会通过 ExecutorService.submit(...) 自行排队运行 问题是我似乎必须为每个作业创建一个单独的 ExecutorService，因为这是判断“作业队列”何时运行的唯一方法' 完整的方法是使用 ExecutorService.awaitTermination(...)。但这似乎效率低下，因为 I can't share a single threadpool作业与其 ExecutorService 之间。

我正在寻找一些替代方案，我正在考虑为每项工作使用 AtomicInteger 。当我提交新任务时增加它，当任务完成时减少它。但随后我必须轮询它何时为零，这看起来很困惑，还有一些异常处理困惑。

看来一定有更好的解决方案？

Answer 1

Submit 返回一个 Future 对象，可用于等待任务完成。您可以跟踪这些并添加一个递归阻塞的方法，直到所有子任务完成。这样您就可以在任何需要的地方重用执行器。

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;

public class JobExecutor {
    ExecutorService executorService = Executors.newFixedThreadPool(1);

    private class Task implements Runnable {
        private final String name;
        private final Task[] subtasks;
        private final ExecutorService executorService;
        private volatile boolean started = false;
        private Future<?> taskFuture;

        // Separate list from subtasks because this is what you'll probably
        // actually use as you may not be passing subtasks as constructor args
        private final List<Task> subtasksToWaitOn = new ArrayList<Task>();

        public Task(String name, ExecutorService executorService,
                Task... subtasks) {
            this.name = name;
            this.executorService = executorService;
            this.subtasks = subtasks;
        }

        public synchronized void start() {
            if (!started) {
                started = true;
                taskFuture = executorService.submit(this);
            }
        }

        public synchronized void blockTillDone() {
            if (started) {
                try {
                    taskFuture.get();
                } catch (InterruptedException e) {
                    // TODO Handle
                } catch (ExecutionException e) {
                    // TODO Handle
                }
                for (Task subtaskToWaitOn : subtasksToWaitOn) {
                    subtaskToWaitOn.blockTillDone();
                }
            } else {
                // TODO throw exception
            }
        }

        @Override
        public void run() {
            for (Task subtask : subtasks) {
                subtask.start();
                subtasksToWaitOn.add(subtask);
            }
            System.out.println("My name is: " + name);
        }
    }

    void testSubmit() {
        Task subsubTask1 = new Task("Subsubtask1", executorService);
        Task subtask1 = new Task("Subtask1", executorService, subsubTask1);
        Task subtask2 = new Task("Subtask2", executorService);
        Task subtask3 = new Task("Subtask3", executorService);
        Task job = new Task("Job", executorService, subtask1, subtask2,
                subtask3);
        job.start();
        job.blockTillDone();
        System.out.println("Job done!");
    }

    public static void main(String[] args) {
        new JobExecutor().testSubmit();
    }
}

打印输出:

My name is: Job
My name is: Subtask1
My name is: Subtask2
My name is: Subtask3
My name is: Subsubtask1
Job done!