只使用

latch = new CountDownLatch(noThreads)

在每个线程中

latch.countDown();

作为屏障

latch.await();

最简单的方法是使用ExecutorService.invokeAll()，它可以在一行程序中执行您想要的操作。用您的话说，您需要修改或包装ComputeDTask以实现Callable<>，这可以为您提供更大的灵活性。可能在您的应用程序中有一个有意义的Callable.call()实现，但如果不使用Executors.callable()，这里有一种包装它的方法。

ExecutorService es = Executors.newFixedThreadPool(2);
List<Callable<Object>> todo = new ArrayList<Callable<Object>>(singleTable.size());

for (DataTable singleTable: uniquePhrases) { 
    todo.add(Executors.callable(new ComputeDTask(singleTable))); 
}

List<Future<Object>> answers = es.invokeAll(todo);

正如其他人指出的那样，如果合适的话，您可以使用invokeAll()的超时版本。在这个例子中，answers将包含一堆返回null的future(参见Executors.callable()的定义)。可能您想做的是稍微重构一下，这样就可以得到一个有用的答案，或者对底层ComputeDTask的引用，但从您的示例中我无法判断。

如果不清楚，请注意，在所有任务完成之前，invokeAll()不会返回。(也就是说，如果被问到，答案集合中的所有future都会报告. isdone()。)这避免了所有的手动关机，等待终止等…并允许您在需要的情况下，在多个周期中巧妙地重用这个ExecutorService。

关于SO有几个相关的问题:

如何等待所有线程完成 从java线程返回值 invokeAll()不愿意接受一个Collection<Callable<t>> 我需要同步吗?

这些都不是严格意义上的问题，但它们确实为人们认为应该如何使用Executor/ExecutorService提供了一点色彩。

我还有一种情况，我有一组文档要爬取。我从一个应该被处理的初始“种子”文档开始，该文档包含到其他也应该被处理的文档的链接，等等。

在我的主程序中，我只想写如下所示的东西，其中Crawler控制了一堆线程。

Crawler c = new Crawler();
c.schedule(seedDocument); 
c.waitUntilCompletion()

如果我想在一棵树上导航，同样的情况也会发生;我将插入根节点，每个节点的处理器将根据需要向队列中添加子节点，一堆线程将处理树中的所有节点，直到没有更多的节点为止。

我在JVM中找不到任何东西，我觉得这有点令人惊讶。所以我写了一个类ThreadPool，一个可以直接使用或子类添加适合域的方法，例如schedule(Document)。希望能有所帮助!

ThreadPool Javadoc | Maven

您可以使用ExecutorService。它将执行所有任务并等待所有线程完成它们的任务。

这是完整的javadoc

您还可以使用此方法的重载版本来指定超时时间。

下面是ExecutorService.invokeAll的示例代码

public class Test {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService service = Executors.newFixedThreadPool(3);
        List<Callable<String>> taskList = new ArrayList<>();
        taskList.add(new Task1());
        taskList.add(new Task2());
        List<Future<String>> results = service.invokeAll(taskList);
        for (Future<String> f : results) {
            System.out.println(f.get());
        }
    }

}

class Task1 implements Callable<String> {
    @Override
    public String call() throws Exception {
        try {
            Thread.sleep(2000);
            return "Task 1 done";
        } catch (Exception e) {
            e.printStackTrace();
            return " error in task1";
        }
    }
}

class Task2 implements Callable<String> {
    @Override
    public String call() throws Exception {
        try {
            Thread.sleep(3000);
            return "Task 2 done";
        } catch (Exception e) {
            e.printStackTrace();
            return " error in task2";
        }
    }
}

有几种方法。

您可以调用第一个ExecutorService。shutdown，然后ExecutorService。返回:

如果该执行程序终止，则为True;如果超时，则为false 在终止之前

So:

有一个叫awaitterminate的函数，但是必须有一个超时 它提供了。这并不能保证当它返回全部时 任务早就完成了。有办法实现这个目标吗?

你只需要在循环中调用awaitterminate。

使用awaitTermination:

这个实现的完整示例:

public class WaitForAllToEnd {

    public static void main(String[] args) throws InterruptedException {
        final int total_threads = 4;
        ExecutorService executor = Executors.newFixedThreadPool(total_threads);
        for(int i = 0; i < total_threads; i++){
            executor.execute(parallelWork(100 + i * 100));
        }

        int count = 0;

        // This is the relevant part
        // Chose the delay most appropriate for your use case
        executor.shutdown();
        while (!executor.awaitTermination(100, TimeUnit.MILLISECONDS)) {
            System.out.println("Waiting "+ count);
            count++;
        }
    }

    private static Runnable parallelWork(long sleepMillis) {
        return () -> {
            try {
                Thread.sleep(sleepMillis);
            } catch (InterruptedException e) {
                // Do Something
            }
            System.out.println("I am Thread : " + Thread.currentThread().getId());
        };
    }
}

使用CountDownLatch:

另一种选择是创建CountDownLatch，其计数等于并行任务的数量。每个线程调用countDownLatch.countDown();，而主线程调用countDownLatch.await();。

这个实现的完整示例:

public class WaitForAllToEnd {

    public static void main(String[] args) throws InterruptedException {
        final int total_threads = 4;
        CountDownLatch countDownLatch = new CountDownLatch(total_threads);
        ExecutorService executor = Executors.newFixedThreadPool(total_threads);
        for(int i = 0; i < total_threads; i++){
            executor.execute(parallelWork(100 + i * 100, countDownLatch));
        }
        countDownLatch.await();
        System.out.println("Exit");
        executor.shutdown();
    }

    private static Runnable parallelWork(long sleepMillis, CountDownLatch countDownLatch) {
        return () -> {
            try {
                Thread.sleep(sleepMillis);
            } catch (InterruptedException e) {
                // Do Something
            }
            System.out.println("I am Thread : " + Thread.currentThread().getId());
            countDownLatch.countDown();
        };
    }
}

使用CyclicBarrier:

另一种方法是使用循环屏障

public class WaitForAllToEnd {

    public static void main(String[] args) throws InterruptedException, BrokenBarrierException {
        final int total_threads = 4;
        CyclicBarrier barrier = new CyclicBarrier(total_threads+ 1);
        ExecutorService executor = Executors.newFixedThreadPool(total_threads);
        for(int i = 0; i < total_threads; i++){
            executor.execute(parallelWork(100 + i * 100, barrier));
        }
        barrier.await();
        System.out.println("Exit");
        executor.shutdown();
    }

    private static Runnable parallelWork(long sleepMillis, CyclicBarrier barrier) {
        return () -> {
            try {
                Thread.sleep(sleepMillis);
            } catch (InterruptedException e) {
                // Do Something
            }
            System.out.println("I am Thread : " + Thread.currentThread().getId());
            try {
                barrier.await();
            } catch (InterruptedException | BrokenBarrierException e) {
              // Do something
            }
        };
    }
}

还有其他方法，但这些方法需要对您的初始需求进行更改，即:

当任务提交时，如何等待所有任务完成 使用ExecutorService.execute()。

你可以在一定的时间间隔内等待任务完成:

int maxSecondsPerComputeDTask = 20;
try {
    while (!es.awaitTermination(uniquePhrases.size() * maxSecondsPerComputeDTask, TimeUnit.SECONDS)) {
        // consider giving up with a 'break' statement under certain conditions
    }
} catch (InterruptedException e) {
    throw new RuntimeException(e);    
}

或者您可以使用ExecutorService.submit(Runnable)并收集它返回的Future对象，然后依次对每个对象调用get()以等待它们完成。

ExecutorService es = Executors.newFixedThreadPool(2);
Collection<Future<?>> futures = new LinkedList<<Future<?>>();
for (DataTable singleTable : uniquePhrases) {
    futures.add(es.submit(new ComputeDTask(singleTable)));
}
for (Future<?> future : futures) {
   try {
       future.get();
   } catch (InterruptedException e) {
       throw new RuntimeException(e);
   } catch (ExecutionException e) {
       throw new RuntimeException(e);
   }
}

正确处理InterruptedException非常重要。它可以让您或库的用户安全地结束一个漫长的进程。