如何检查在执行器服务上运行的所有任务是否都已完成

2022-09-01 02:57:38

我有ConcurrentLinkedDeque,我将其用于同步推送/弹出元素,并且我有一些异步任务,这些任务从堆栈中获取一个元素,如果该元素具有邻居,则将其推送到堆栈。

示例代码:

private ConcurrentLinkedDeque<Item> stack = new ConcurrentLinkedDeque<>();
private ExecutorService exec = Executors.newFixedThreadPool(5);

    while ((item = stack.pollFirst()) != null) {
                if (item == null) {
                } else {
                    Runnable worker = new Solider(this, item);
                    exec.execute(worker);
                }
            }

   class Solider{
         public void run(){
             if(item.hasNeighbors){
                for(Item item:item.neighbors){
                    stack.push(item)
                }
             } 
         }
    }

我想在while循环中有额外的语句来回答这个问题 - “Executor中的任何任务都在工作?


答案 1

没有一种干净的方法来检查是否所有Runnables都已完成,如果您使用.除非你在Runnable本身中建立一个机制来做到这一点(在我看来这很草率)。ExecutorService.execute(Runnable)

请改用:
使用 .此方法将返回 一个,它是 结果的句柄。使用 Futures 提供了一种检查结果的干净方法。ExecutorService.submit(Runnable)Future<?>Runnable

您所要做的就是维护您提交的期货列表,然后您可以迭代整个期货列表,然后:
A)等待所有期货以阻塞方式完成或
B)检查所有期货是否以非阻塞方式完成。

下面是一个代码示例:

List<Future<?>> futures = new ArrayList<Future<?>>();
ExecutorService exec = Executors.newFixedThreadPool(5);

// Instead of using exec.execute() use exec.submit()
// because it returns a monitorable future
while((item = stack.pollFirst()) != null){
    Runnable worker = new Solider(this, item);
    Future<?> f = exec.submit(worker);
    futures.add(f);
}

// A) Await all runnables to be done (blocking)
for(Future<?> future : futures)
    future.get(); // get will block until the future is done

// B) Check if all runnables are done (non-blocking)
boolean allDone = true;
for(Future<?> future : futures){
    allDone &= future.isDone(); // check if future is done
}

答案 2

更新:使用Java 8 +CompletableFutures,您可以使用其新的回调函数来管理它。首先,您需要创建所需的所有CompletableFutures,这些功能也将开始运行,例如:

我们需要累积数组中生成的所有期货,以便稍后将它们传递给ComppletableFuture.allOf(CompletableFutures...)

因此,假设您有一个想要异步计算其生日前天数的人员列表:

首先,我们创建所有这些需要的未来,并将它们收集在一个数组中:

CompletableFuture<?>[] completables = people.stream()
    .map(p -> createCompletableFuture(p))
    .toArray(CompletableFuture<?>[]::new);

private CompletableFuture createCompletableFuture(Person p) {
        return CompletableFuture.runAsync(daysUntillBirthday(p));
    }

然后,您将这些可满足性传递给新的可填料未来:

CompletableFuture c = CompletableFuture.allOf(completables)

现在,您可以使用以下命令检查是否仍有期货在运行:

c.isDone()