我需要一次执行一定数量的任务4,就像这样:

ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
while(...) {
    taskExecutor.execute(new MyTask());
}
//...wait for completion somehow

当所有这些都完成后,我如何得到通知?现在我想不出比设置一些全局任务计数器更好的方法,并在每个任务结束时减少它,然后在无限循环中监视这个计数器变成0;或获取一个期货列表,并在无限循环监视器isDone为所有它们。不涉及无限循环的更好的解决方案是什么?

谢谢。


当前回答

Java 5及以后版本中的CyclicBarrier类就是为这类事情而设计的。

其他回答

只是在这里提供更多不同于使用闩锁/屏障的选择。 你也可以在它们全部使用完CompletionService之前得到部分结果。

来自Java并发实践: “如果您有一批计算要提交给Executor,并且您希望检索它们的结果 可用时,您可以保留与每个任务关联的Future,并通过调用get来重复轮询完成 超时为0。这是可能的,但很乏味。幸运的是,还有更好的方法:完井服务。”

这里是实现

public class TaskSubmiter {
    private final ExecutorService executor;
    TaskSubmiter(ExecutorService executor) { this.executor = executor; }
    void doSomethingLarge(AnySourceClass source) {
        final List<InterestedResult> info = doPartialAsyncProcess(source);
        CompletionService<PartialResult> completionService = new ExecutorCompletionService<PartialResult>(executor);
        for (final InterestedResult interestedResultItem : info)
            completionService.submit(new Callable<PartialResult>() {
                public PartialResult call() {
                    return InterestedResult.doAnOperationToGetPartialResult();
                }
        });

    try {
        for (int t = 0, n = info.size(); t < n; t++) {
            Future<PartialResult> f = completionService.take();
            PartialResult PartialResult = f.get();
            processThisSegment(PartialResult);
            }
        } 
        catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } 
        catch (ExecutionException e) {
            throw somethinghrowable(e.getCause());
        }
    }
}

ExecutorService.invokeAll()为您完成。

ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
List<Callable<?>> tasks; // your tasks
// invokeAll() returns when all tasks are complete
List<Future<?>> futures = taskExecutor.invokeAll(tasks);

您可以使用自己的ExecutorCompletionService子类来包装taskExecutor,并使用自己的BlockingQueue实现在每个任务完成时获得通知,并在完成的任务数量达到预期目标时执行任何回调或其他操作。

所以我把我的答案贴在这里,以防有人想要一个更简单的方法来做到这一点

ExecutorService executor = Executors.newFixedThreadPool(10);
CompletableFuture[] futures = new CompletableFuture[10];
int i = 0;
while (...) {
    futures[i++] =  CompletableFuture.runAsync(runner, executor);
}

CompletableFuture.allOf(futures).join(); // THis will wait until all future ready.

遵循以下方法之一。

遍历从ExecutorService上的submit返回的所有Future任务,并按照Kiran的建议使用Future对象上的阻塞调用get()检查状态 在ExecutorService上使用invokeAll() CountDownLatch ForkJoinPool或Executors.html#newWorkStealingPool 按照适当的顺序使用ThreadPoolExecutor的shutdown、awaitterminate、shutdownNow api

相关的SE问题:

CountDownLatch是如何在Java多线程中使用的?

如何正确关闭java ExecutorService