Wie kann man mit ExecutorService warten, bis alle Threads beendet sind?

Question

Ich muss eine gewisse Anzahl von Aufgaben 4 auf einmal ausführen, etwa so:

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

Wie kann ich benachrichtigt werden, wenn sie alle abgeschlossen sind? Im Moment fällt mir nichts Besseres ein, als einen globalen Aufgabenzähler zu setzen und ihn am Ende jeder Aufgabe zu verringern und dann in einer Endlosschleife zu überwachen, dass dieser Zähler 0 wird; oder eine Liste von Futures zu erhalten und in einer Endlosschleife isDone für alle zu überwachen. Was sind bessere Lösungen, die keine Endlosschleifen beinhalten?

Danke.

Answer 1

Nur meine Meinung dazu. Um das Erfordernis der CountDownLatch um die Anzahl der Aufgaben im Voraus zu wissen, könnten Sie dies auf die alte Art und Weise tun, indem Sie eine einfache Semaphore .

ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
int numberOfTasks=0;
Semaphore s=new Semaphore(0);
while(...) {
    taskExecutor.execute(new MyTask());
    numberOfTasks++;
}

try {
    s.aquire(numberOfTasks);
...

Rufen Sie in Ihrer Aufgabe einfach s.release() wie Sie es tun würden latch.countDown();

Answer 2

Ein bisschen spät, aber der Vollständigkeit halber...

Anstatt zu warten, bis alle Aufgaben erledigt sind, können Sie nach dem Hollywood-Prinzip vorgehen: "Ruf mich nicht an, ich rufe dich an" - wenn ich fertig bin. Ich denke, der resultierende Code ist eleganter...

Guava bietet dazu einige interessante Werkzeuge.

Ein Beispiel:

Einen ExecutorService in einen ListeningExecutorService verpacken:

ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));

Eine Sammlung von Callables zur Ausführung vorlegen ::

for (Callable<Integer> callable : callables) {
  ListenableFuture<Integer> lf = service.submit(callable);
  // listenableFutures is a collection
  listenableFutures.add(lf)
});

Nun zum Wesentlichen:

ListenableFuture<List<Integer>> lf = Futures.successfulAsList(listenableFutures);

Fügen Sie dem ListenableFuture einen Callback hinzu, mit dem Sie benachrichtigt werden können, wenn alle Futures abgeschlossen sind:

Futures.addCallback(lf, new FutureCallback<List<Integer>> () {
    @Override
    public void onSuccess(List<Integer> result) {
        // do something with all the results
    }

    @Override
    public void onFailure(Throwable t) {
        // log failure
    }
});

Dies hat auch den Vorteil, dass Sie alle Ergebnisse an einem Ort sammeln können, sobald die Bearbeitung abgeschlossen ist...

Weitere Informationen aquí

Answer 3

En CyclicBarrier Klasse in Java 5 und höher ist für diese Art von Aufgaben konzipiert.

Answer 4

Hier sind zwei Optionen, nur ein bisschen verwirrt, welche ist am besten zu gehen.

Option 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();

Option 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();

Hier setzen future.get(); in try catch ist eine gute Idee richtig?

Answer 5

Folgen Sie einem der folgenden Ansätze.

1. Iterieren Sie durch alle Zukunft Aufgaben, zurückgegeben von submit において ExecutorService und überprüfen Sie den Status mit dem blockierenden Aufruf get() において Future Objekt, wie es von Kiran
2. Utilisez invokeAll() において ExecutorService
3. CountDownLatch
4. ForkJoinPool o Executors.html#newWorkStealingPool
5. Utilisez shutdown, awaitTermination, shutdownNow APIs des ThreadPoolExecutor in der richtigen Reihenfolge

Verwandte SE-Fragen:

Wie wird CountDownLatch in Java Multithreading verwendet?

Wie man den Java ExecutorService richtig herunterfährt