我需要一次执行一定数量的任务4，就像这样:
ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
while(...) {
taskExecutor.execute(new MyTask());
}
//...wait for completion somehow
当所有这些都完成后，我如何得到通知?现在我想不出比设置一些全局任务计数器更好的方法，并在每个任务结束时减少它，然后在无限循环中监视这个计数器变成0;或获取一个期货列表，并在无限循环监视器isDone为所有它们。不涉及无限循环的更好的解决方案是什么?
谢谢。
当前回答
遵循以下方法之一。
遍历从ExecutorService上的submit返回的所有Future任务，并按照Kiran的建议使用Future对象上的阻塞调用get()检查状态 在ExecutorService上使用invokeAll() CountDownLatch ForkJoinPool或Executors.html#newWorkStealingPool 按照适当的顺序使用ThreadPoolExecutor的shutdown、awaitterminate、shutdownNow api
相关的SE问题:
CountDownLatch是如何在Java多线程中使用的?
如何正确关闭java ExecutorService
其他回答
只是在这里提供更多不同于使用闩锁/屏障的选择。 你也可以在它们全部使用完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());
}
}
}
在执行器getActiveCount() -中有一个方法可以给出活动线程的计数。
在跨越线程之后，我们可以检查activeCount()值是否为0。一旦该值为零，就意味着当前没有活动线程在运行，这意味着任务已经完成:
while (true) {
if (executor.getActiveCount() == 0) {
//ur own piece of code
break;
}
}
这里有两个选择，只是有点困惑，哪个是最好的。
选项1:
ExecutorService es = Executors.newFixedThreadPool(4);
List<Runnable> tasks = getTasks();
CompletableFuture<?>[] futures = tasks.stream()
.map(task -> CompletableFuture.runAsync(task, es))
.toArray(CompletableFuture[]::new);
CompletableFuture.allOf(futures).join();
es.shutdown();
选项2:
ExecutorService es = Executors.newFixedThreadPool(4);
List< Future<?>> futures = new ArrayList<>();
for(Runnable task : taskList) {
futures.add(es.submit(task));
}
for(Future<?> future : futures) {
try {
future.get();
}catch(Exception e){
// do logging and nothing else
}
}
es.shutdown();
这里放入future.get();试抓是个好主意，对吧?
你可以使用下面的代码:
public class MyTask implements Runnable {
private CountDownLatch countDownLatch;
public MyTask(CountDownLatch countDownLatch {
this.countDownLatch = countDownLatch;
}
@Override
public void run() {
try {
//Do somethings
//
this.countDownLatch.countDown();//important
} catch (InterruptedException ex) {
Thread.currentThread().interrupt();
}
}
}
CountDownLatch countDownLatch = new CountDownLatch(NUMBER_OF_TASKS);
ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
for (int i = 0; i < NUMBER_OF_TASKS; i++){
taskExecutor.execute(new MyTask(countDownLatch));
}
countDownLatch.await();
System.out.println("Finish tasks");
在Java8中，你可以用CompletableFuture来实现:
ExecutorService es = Executors.newFixedThreadPool(4);
List<Runnable> tasks = getTasks();
CompletableFuture<?>[] futures = tasks.stream()
.map(task -> CompletableFuture.runAsync(task, es))
.toArray(CompletableFuture[]::new);
CompletableFuture.allOf(futures).join();
es.shutdown();
