如何在不关闭 Executor 的情况下等待 ThreadPoolExecutor 中的所有任务完成？

Question

我不能使用shutdown() 和awaitTermination()，因为它可能会在等待时将新任务添加到 ThreadPoolExecutor。

所以我正在寻找一种方法，等待 ThreadPoolExecutor 清空它的队列并完成它的所有任务，而不停止在此之前添加新任务。

如果有什么不同，这适用于 Android。

谢谢

更新：很多周后，我重新审视了这个，发现修改后的 CountDownLatch 在这种情况下更适合我。我会保留答案，因为它更适用于我的问题。

Answer 1

如果您想知道某项任务或某批任务何时完成，您可以使用ExecutorService.submit(Runnable)。调用此方法会返回一个 Future 对象，该对象可以放入一个 Collection 中，然后您的主线程将遍历每个对象调用 Future.get()。这将导致您的主线程暂停执行，直到 ExecutorService 处理完所有 Runnable 任务。

Collection<Future<?>> futures = new LinkedList<Future<?>>();
futures.add(executorService.submit(myRunnable));
for (Future<?> future:futures) {
    future.get();
}

Answer 2

我的场景是一个网络爬虫，用于从网站获取一些信息，然后对其进行处理。一个 ThreadPoolExecutor 用于加快进程，因为可以同时加载许多页面。因此，新任务将在现有任务中创建，因为爬虫将跟踪每个页面中的超链接。问题是一样的：主线程不知道什么时候所有的任务都完成了，它可以开始处理结果。我用一种简单的方法来确定这一点。它不是很优雅，但在我的情况下有效：

while (executor.getTaskCount()!=executor.getCompletedTaskCount()){
    System.err.println("count="+executor.getTaskCount()+","+executor.getCompletedTaskCount());
    Thread.sleep(5000);
}
executor.shutdown();
executor.awaitTermination(60, TimeUnit.SECONDS);

Answer 3

也许您正在寻找CompletionService 来管理批量任务，另请参阅this answer。

Answer 4

（这是尝试通过我自己的调整来重现 Thilo 之前删除的答案。）

我认为您可能需要澄清您的问题，因为有一个隐含的无限条件......在某些时候您必须决定关闭您的执行程序，此时它将不再接受任何任务。您的问题似乎暗示您想等到知道不会提交更多任务，您只能在自己的应用程序代码中知道。

以下答案将允许您顺利过渡到新 TPE（无论出于何种原因），完成所有当前提交的任务，并且不会拒绝新 TPE 的新任务。它可能会回答你的问题。 @Thilo 也可能。

假设您已经在某处定义了一个可见的 TPE：

AtomicReference<ThreadPoolExecutor> publiclyAvailableTPE = ...;

然后您可以这样编写 TPE 交换例程。也可以使用同步方法编写，但我认为这更简单：

void rotateTPE()
{
   ThreadPoolExecutor newTPE = createNewTPE();
   // atomic swap with publicly-visible TPE
   ThreadPoolExecutor oldTPE = publiclyAvailableTPE.getAndSet(newTPE);
   oldTPE.shutdown();

   // and if you want this method to block awaiting completion of old tasks in  
   // the previously visible TPE
   oldTPE.awaitTermination();
} 

或者，如果您真的不想杀死线程池，那么您的提交者端将需要在某个时候处理被拒绝的任务，您可以将null 用于新的 TPE：

void killTPE()
{
   ThreadPoolExecutor oldTPE = publiclyAvailableTPE.getAndSet(null);
   oldTPE.shutdown();

   // and if you want this method to block awaiting completion of old tasks in  
   // the previously visible TPE
   oldTPE.awaitTermination();
} 

这可能会导致上游问题，调用者需要知道如何处理null。

您也可以用一个简单地拒绝每个新执行的虚拟 TPE 进行替换，但这相当于您在 TPE 上调用 shutdown() 时发生的情况。

Answer 5

如果您不想使用shutdown，请遵循以下方法：

1. 从ExecutorService 上的提交遍历所有Future 任务，并按照Tim Bender 的建议在Future 对象上使用阻塞调用get() 检查状态

2. 使用其中一种

   1. 在ExecutorService 上使用invokeAll
   2. 使用CountDownLatch
   3. 使用ForkJoinPool 或newWorkStealingPool 的Executors（从java 8 开始）

executor服务上的invokeAll()也达到CountDownLatch的相同目的

相关的 SE 问题：

How to wait for a number of threads to complete?

Answer 6

您可以在 Runner 类上调用 waitTillDone()：

Runner runner = Runner.runner(10);

runner.runIn(2, SECONDS, runnable);
runner.run(runnable); // each of this runnables could submit more tasks

runner.waitTillDone(); // blocks until all tasks are finished (or failed)

// and now reuse it

runner.runIn(500, MILLISECONDS, callable);

runner.waitTillDone();
runner.shutdown();

要使用它，请将此 gradle/maven 依赖项添加到您的项目中：'com.github.matejtymes:javafixes:1.0'

更多详情请看这里：https://github.com/MatejTymes/JavaFixes 或这里：http://matejtymes.blogspot.com/2016/04/executor-that-notifies-you-when-task.html

Answer 7

尝试使用队列大小和活动任务计数，如下所示

 while (executor.getThreadPoolExecutor().getActiveCount() != 0 || !executor.getThreadPoolExecutor().getQueue().isEmpty()){
                     try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
            }
        }