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

Sie könnten Ihre Aufgaben in eine andere Runnable einpacken, die Benachrichtigungen sendet:

taskExecutor.execute(new Runnable() {
  public void run() {
    taskStartedNotification();
    new MyTask().run();
    taskFinishedNotification();
  }
});

Answer 2

Sauberer Weg mit ExecutorService

 List<Future<Void>> results = null;
 try {
     List<Callable<Void>> tasks = new ArrayList<>();
     ExecutorService executorService = Executors.newFixedThreadPool(4);
     results = executorService.invokeAll(tasks);
 } catch (InterruptedException ex) {
     ...
 } catch (Exception ex) {
     ...
 }

Answer 3

Versuch-mit-Ressourcen-Syntax auf AutoCloseable Testamentsvollstreckerdienst mit Projekt Webstuhl

Projekt Webstuhl zielt darauf ab, die Gleichzeitigkeitsfähigkeiten in Java um neue Funktionen zu erweitern.

Eines dieser Merkmale besteht darin, dass die ExecutorService AutoCloseable . Dies bedeutet, dass jeder ExecutorService Umsetzung wird eine [close](https://download.java.net/java/early_access/loom/docs/api/java.base/java/lang/AutoCloseable.html#close()) Methode. Und das bedeutet, dass wir die Versuch-mit-Ressourcen Syntax zum automatischen Schließen eines ExecutorService Objekt.

En [ExecutorService#close](https://download.java.net/java/early_access/loom/docs/api/java.base/java/util/concurrent/ExecutorService.html#close()) Methode blockiert, bis alle eingereichten Aufgaben abgeschlossen sind. Verwendung von close ersetzt den Aufruf von shutdown & awaitTermination .

Sein AutoCloseable trägt zu dem Versuch von Project Loom bei, die "strukturierte Gleichzeitigkeit" zu Java.

try (
    ExecutorService executorService = Executors.… ;
) {
    // Submit your `Runnable`/`Callable` tasks to the executor service.
    …
}
// At this point, flow-of-control blocks until all submitted tasks are done/canceled/failed.
// After this point, the executor service will have been automatically shutdown, wia `close` method called by try-with-resources syntax.

Für weitere Informationen über Project Loom suchen Sie nach Vorträgen und Interviews von Ron Pressler und anderen Mitgliedern des Project Loom-Teams. Konzentrieren Sie sich auf die jüngeren Beiträge, da sich Project Loom weiterentwickelt hat.

Experimentelle Versionen der Project Loom-Technologie sind jetzt verfügbar auf der Grundlage des frühzeitigen Zugangs Java 18 .

Answer 4

Nur um mehr Alternativen zu bieten, können Sie hier verschiedene Verriegelungen/Sperren verwenden. Sie können auch die Teilergebnisse erhalten, bis alle von ihnen mit CompletionService .

Aus Java Gleichzeitigkeit in der Praxis: "Wenn Sie einen Stapel von Berechnungen an einen Executor übergeben und die Ergebnisse abrufen möchten, sobald sie abrufen möchten, können Sie die mit jeder Aufgabe verknüpfte Zukunft beibehalten und wiederholt nach dem Abschluss fragen, indem Sie get mit einem Timeout von Null aufrufen. Dies ist möglich, aber mühsam . Zum Glück gibt es eine bessere Art einen Abschlussdienst".

Hier die Umsetzung

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

Answer 5

Ich habe gerade ein Beispielprogramm geschrieben, das Ihr Problem löst. Es wurde keine prägnante Implementierung angegeben, also füge ich eine hinzu. Sie können zwar executor.shutdown() y executor.awaitTermination() ist dies nicht die beste Praxis, da die Zeit, die die verschiedenen Threads benötigen, nicht vorhersehbar ist.

ExecutorService es = Executors.newCachedThreadPool();
    List<Callable<Integer>> tasks = new ArrayList<>();

    for (int j = 1; j <= 10; j++) {
        tasks.add(new Callable<Integer>() {

            @Override
            public Integer call() throws Exception {
                int sum = 0;
                System.out.println("Starting Thread "
                        + Thread.currentThread().getId());

                for (int i = 0; i < 1000000; i++) {
                    sum += i;
                }

                System.out.println("Stopping Thread "
                        + Thread.currentThread().getId());
                return sum;
            }

        });
    }

    try {
        List<Future<Integer>> futures = es.invokeAll(tasks);
        int flag = 0;

        for (Future<Integer> f : futures) {
            Integer res = f.get();
            System.out.println("Sum: " + res);
            if (!f.isDone()) 
                flag = 1;
        }

        if (flag == 0)
            System.out.println("SUCCESS");
        else
            System.out.println("FAILED");

    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }