Java 执行器：等待任务终止。 [复制]

Question

我需要提交一些任务，然后等待它们，直到所有结果都可用。他们每个人都将String 添加到Vector（默认情况下是同步的）。然后我需要为 Vector 中的每个结果启动一个新任务，但只有在所有先前的任务都停止工作时我才需要这样做。

我想使用 Java Executor，特别是我尝试使用 Executors.newFixedThreadPool(100) 以使用固定数量的线程（我有可变数量的任务，可以是 10 或 500）但我是新的 executors 和我不知道如何等待任务终止。 这类似于我的程序需要执行的伪代码：

ExecutorService e = Executors.newFixedThreadPool(100);
while(true){

/*do something*/

for(...){
<start task>
}

<wait for all task termination>

for each String in result{
<start task>
}

<wait for all task termination>
}

我无法执行 e.shutdown，因为我有一段时间（真的），我需要重用 executorService...

你能帮帮我吗？你能给我推荐一本关于 java executors 的指南/书吗？

Answer 1

你能给我推荐一本关于 java执行器？？

这部分我可以回答：

Java Concurrency in Practice by Brian Goetz（与 Tim Peierls、Joshua Bloch、Joseph Bowbeer、David Holmes 和 Doug Lea）很可能是您的最佳选择。

这不仅仅是只关于执行器，而是涵盖了java.util.concurrent 包的一般性，以及基本的并发概念和技术，以及一些高级主题，例如 Java 内存模型。

Answer 2

ExecutorService 为您提供了一种机制，可以同时执行多个任务并返回 Future 对象的集合（表示任务的异步计算）。

Collection<Callable<?>> tasks = new LinkedList<Callable<?>>();
//populate tasks
for (Future<?> f : executorService.invokeAll(tasks)) { //invokeAll() blocks until ALL tasks submitted to executor complete
    f.get(); 
}

如果您有Runnables 而不是Callables，则可以使用以下方法轻松地将Runnable 转换为Callable<Object>：

Callable<?> c = Executors.callable(runnable);

Answer 3

ExecutorService executor = ...
//submit tasks
executor.shutdown(); // previously submitted tasks are executed, 
                     // but no new tasks will be accepted
while(!executor.awaitTermination(1, TimeUnit.SECONDS))
    ;

如果不创建自定义 ExecutorService，就没有简单的方法来做你想做的事。

Answer 4

而不是直接提交Runnables 或Callables 到Executor 并存储相应的Future 返回值我建议使用CompletionService 实现来检索每个Future when它完成了。这种方法将任务的生产与已完成任务的消耗分离，例如，允许在一段时间内在生产者线程上发起新任务。

Collection<Callable<Result>> workItems = ...
ExecutorService executor = Executors.newSingleThreadExecutor();
CompletionService<Result> compService = new ExecutorCompletionService<Result>(executor);

// Add work items to Executor.
for (Callable<Result> workItem : workItems) {
  compService.submit(workItem);
}

// Consume results as they complete (this would typically occur on a different thread).
for (int i=0; i<workItems.size(); ++i) {
  Future<Result> fut = compService.take(); // Will block until a result is available.
  Result result = fut.get(); // Extract result; this will not block.
}

Answer 5

当您提交到执行器服务时，您会得到一个 Future 对象。

将这些对象存储在一个集合中，然后依次调用每个对象上的get()。 get() 阻塞，直到底层作业完成，因此结果是在所有底层作业完成后对每个调用 get() 将完成。

例如

Collection<Future> futures = ...
for (Future f : futures) {
   Object result = f.get();
   // maybe do something with the result. This could be a
   // genericised Future<T>
}
System.out.println("Tasks completed");

所有这些都完成后，开始您的第二次提交。请注意，这可能不是对线程池的最佳使用，因为它将变为休眠状态，然后您将重新填充它。如果可能的话，试着让它忙着做事。