等待ExecutorService所有任务完成的最简单方法是什么?我的任务主要是计算性的,所以我只想运行大量的作业——每个核心上都有一个。现在我的设置是这样的:

ExecutorService es = Executors.newFixedThreadPool(2);
for (DataTable singleTable : uniquePhrases) {   
    es.execute(new ComputeDTask(singleTable));
}
try{
    es.wait();
} 
catch (InterruptedException e){
    e.printStackTrace();
}

ComputeDTask实现了runnable。这似乎正确地执行了任务,但代码在wait()时崩溃,并出现IllegalMonitorStateException。这是奇怪的,因为我玩了一些玩具的例子,它似乎工作。

uniquePhrases包含数万个元素。我应该用另一种方法吗?我在寻找一些尽可能简单的东西


当前回答

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

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

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

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

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

ThreadPool Javadoc | Maven

其他回答

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

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非常重要。它可以让您或库的用户安全地结束一个漫长的进程。

如果您希望等待所有任务完成,请使用关机方法而不是等待。然后在后面加上awaitterminate。

此外,您还可以使用Runtime。availableProcessors来获取硬件线程的数量,以便正确地初始化线程池。

我将等待执行程序以您认为适合任务完成的指定超时终止。

 try {  
         //do stuff here 
         exe.execute(thread);
    } finally {
        exe.shutdown();
    }
    boolean result = exe.awaitTermination(4, TimeUnit.HOURS);
    if (!result)

    {
        LOGGER.error("It took more than 4 hour for the executor to stop, this shouldn't be the normal behaviour.");
    }

最简单的方法是使用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提供了一点色彩。

听起来好像你需要ForkJoinPool并使用全局池来执行任务。

public static void main(String[] args) {
    // the default `commonPool` should be sufficient for many cases.
    ForkJoinPool pool = ForkJoinPool.commonPool(); 
    // The root of your task that may spawn other tasks. 
    // Make sure it submits the additional tasks to the same executor that it is in.
    Runnable rootTask = new YourTask(pool); 
    pool.execute(rootTask);
    pool.awaitQuiescence(...);
    // that's it.
}

美在泳池里。awaitQuiescence方法将阻塞利用调用者的线程来执行它的任务,然后当它真的为空时返回。