ExecutorService, wie Sie warten, bis alle Aufgaben abgeschlossen sind

196

Was ist der einfachste Weg, um zu warten, bis alle Aufgaben erledigt sind ExecutorService? Meine Aufgabe ist in erster Linie rechnerisch, daher möchte ich nur eine große Anzahl von Jobs ausführen - einen auf jedem Kern. Im Moment sieht mein Setup so aus:

ExecutorService es = Executors.newFixedThreadPool(2);
for (DataTable singleTable : uniquePhrases) {   
    es.execute(new ComputeDTask(singleTable));
}
try{
    es.wait();
} 
catch (InterruptedException e){
    e.printStackTrace();
}

ComputeDTaskimplementiert lauffähig. Dies scheint die Aufgaben korrekt auszuführen, aber der Code stürzt wait()mit ab IllegalMonitorStateException. Das ist seltsam, weil ich mit einigen Spielzeugbeispielen herumgespielt habe und es schien zu funktionieren.

uniquePhrasesenthält mehrere Zehntausende von Elementen. Sollte ich eine andere Methode verwenden? Ich suche etwas so Einfaches wie möglich

George Smiley
quelle
1
Wenn Sie wait verwenden möchten (die Antworten sagen Ihnen, dass Sie dies nicht tun): Sie müssen immer auf dem Objekt synchronisieren (in diesem Fall es), wenn Sie darauf warten möchten - die Sperre wird während des Wartens automatisch aufgehoben
mihi
13
Ein besserer Weg, um den ThreadPool zu initialisieren, istExecutors.newFixedThreadPool(System.getRuntime().availableProcessors());
7
Runtime.getRuntime (). AvailableProcessors ();
Vik Gamov
[Dies] [1] ist eine interessante Alternative. [1]: stackoverflow.com/questions/1250643/…
Răzvan Petruescu
1
Wenn Sie CountDownLatch verwenden möchten, ist dies Beispielcode: stackoverflow.com/a/44127101/4069305
Tuan Pham

Antworten:

213

Der einfachste Ansatz besteht darin, zu verwenden ExecutorService.invokeAll(), was in einem Einzeiler das tut, was Sie wollen. In Ihrer Sprache müssen Sie ComputeDTaskdie Implementierung ändern oder umbrechen Callable<>, was Ihnen einiges an Flexibilität geben kann. Wahrscheinlich gibt es in Ihrer App eine sinnvolle Implementierung von Callable.call(), aber hier ist eine Möglichkeit, sie zu verpacken, wenn Sie sie nicht verwenden Executors.callable().

ExecutorService es = Executors.newFixedThreadPool(2);
List<Callable<Object>> todo = new ArrayList<Callable<Object>>(singleTable.size());

for (DataTable singleTable: uniquePhrases) { 
    todo.add(Executors.callable(new ComputeDTask(singleTable))); 
}

List<Future<Object>> answers = es.invokeAll(todo);

Wie andere bereits betont haben, können Sie invokeAll()gegebenenfalls die Timeout-Version von verwenden . In diesem Beispiel answerswird es eine Reihe von Futures enthalten, die Nullen zurückgeben (siehe Definition von Executors.callable(). Wahrscheinlich möchten Sie ein leichtes Refactoring durchführen, damit Sie eine nützliche Antwort oder einen Verweis auf den zugrunde liegenden Wert erhalten ComputeDTask, aber ich kann Erzählen Sie nicht von Ihrem Beispiel.

Wenn es nicht klar ist, beachten Sie, dass invokeAll()es nicht zurückkehrt, bis alle Aufgaben abgeschlossen sind. (Das heißt, alle Futures in Ihrer answersSammlung werden gemeldet, .isDone()wenn Sie dazu aufgefordert werden.) Dies vermeidet das manuelle Herunterfahren, Warten auf Beendigung usw. und ermöglicht es Ihnen, dies bei ExecutorServiceBedarf für mehrere Zyklen sauber wiederzuverwenden .

Es gibt einige verwandte Fragen zu SO:

Keines davon ist genau auf Ihre Frage zugeschnitten, aber es liefert ein wenig Farbe darüber, wie Leute denken Executor/ verwendet ExecutorServicewerden sollten.

andersoj
quelle
9
Dies ist perfekt, wenn Sie alle Ihre Jobs in einem Stapel hinzufügen und an der Liste der Callables festhalten. Es funktioniert jedoch nicht, wenn Sie ExecutorService.submit () in einer Rückruf- oder Ereignisschleifensituation aufrufen.
Desty
2
Ich denke, es ist erwähnenswert, dass shutdown () immer noch aufgerufen werden sollte, wenn der ExecutorService nicht mehr benötigt wird, da sonst die Threads niemals beendet werden (außer in den Fällen, in denen corePoolSize = 0 oder allowCoreThreadTimeOut = true ist).
John29
tolle! Genau das, wonach ich gesucht habe. Vielen Dank für die Antwort. Lass mich das ausprobieren.
MohamedSanaulla
59

Wenn Sie warten möchten, bis alle Aufgaben abgeschlossen sind, verwenden Sie shutdownstattdessen die Methode wait. Dann folgen Sie ihm mit awaitTermination.

Außerdem können Sie Runtime.availableProcessorsdie Anzahl der Hardware-Threads abrufen, um Ihren Threadpool ordnungsgemäß zu initialisieren.

NG.
quelle
27
shutdown () verhindert, dass der ExecutorService neue Aufgaben akzeptiert, und schließt inaktive Worker-Threads. Es ist nicht angegeben, auf den Abschluss des Herunterfahrens zu warten, und die Implementierung in ThreadPoolExecutor wartet nicht.
Alain O'Dea
1
@Alain - danke. Ich hätte awaitTermination erwähnen sollen. Fest.
NG.
5
Was ist, wenn eine Aufgabe zum Beenden weitere Aufgaben planen muss? Sie können beispielsweise eine Multithread-Baumdurchquerung durchführen, bei der Zweige an Arbeitsthreads übergeben werden. In diesem Fall akzeptiert der ExecutorService keine rekursiv geplanten Jobs, da er sofort heruntergefahren wird.
Brian Gordon
2
awaitTerminationerfordert eine Timeout-Zeit als Parameter. Während es möglich ist, eine endliche Zeit anzugeben und eine Schleife darum zu legen, um zu warten, bis alle Threads fertig sind, habe ich mich gefragt, ob es eine elegantere Lösung gibt.
Abhishek S
1
Sie haben Recht, aber sehen Sie diese Antwort - stackoverflow.com/a/1250655/263895 - Sie könnten ihm immer eine unglaublich lange Auszeit geben
NG.
48

Wenn das Warten auf das ExecutorServiceBeenden aller Aufgaben im Ziel nicht genau Ihr Ziel ist, sondern das Warten, bis ein bestimmter Stapel von Aufgaben abgeschlossen ist, können Sie ein CompletionService- speziell ein ExecutorCompletionService.

Die Idee ist, eine ExecutorCompletionServiceVerpackung zu erstellen Executor, einreichen einige bekannte Anzahl von Aufgaben durch die CompletionService, dann ist das Zeichnen derselben Anzahl von Ergebnissen aus dem Abschluss - Warteschlange unter Verwendung von entweder take()(die Blöcke) oder poll()(was nicht). Sobald Sie alle erwarteten Ergebnisse für die von Ihnen eingereichten Aufgaben gezeichnet haben, wissen Sie, dass alle erledigt sind.

Lassen Sie mich dies noch einmal sagen, da es aus der Benutzeroberfläche nicht ersichtlich ist: Sie müssen wissen, wie viele Dinge Sie in das CompletionServiceeingeben, um zu wissen, wie viele Dinge Sie herausziehen möchten. Dies ist besonders bei der take()Methode von Bedeutung: Rufen Sie sie einmal zu oft auf, und Ihr aufrufender Thread wird blockiert, bis ein anderer Thread einen anderen Job an denselben sendet CompletionService.

Es gibt einige Beispiele, die zeigen, wieCompletionService das Buch Java Concurrency in der Praxis verwendet wird .

seh
quelle
Dies ist ein guter Kontrapunkt zu meiner Antwort - ich würde sagen, die einfache Antwort auf die Frage lautet invokeAll (); aber @seh hat es richtig, wenn Gruppen von Jobs an die ES gesendet werden und darauf warten, dass sie abgeschlossen sind ... --JA
andersoj
@ om-nom-nom, danke für die Aktualisierung der Links. Ich bin froh zu sehen, dass die Antwort immer noch nützlich ist.
seh
1
Gute Antwort, ich wusste nichtCompletionService
Vic
1
Dies ist der zu verwendende Ansatz, wenn Sie einen vorhandenen ExecutorService nicht herunterfahren möchten, sondern nur einen Stapel von Aufgaben senden möchten und wissen möchten, wann alle abgeschlossen sind.
ToolmakerSteve
11

Wenn Sie warten möchten, bis der Executor-Service die Ausführung beendet hat, rufen Sie an und warten Sie shutdown()dann auf Terminierung (Einheiten, Einheitentyp) , z awaitTermination(1, MINUTE). Der ExecutorService blockiert nicht auf seinem eigenen Monitor, sodass Sie waitusw. nicht verwenden können .

mdma
quelle
Ich denke, es wartet auf die Beendigung.
NG.
@SB - Danke - ich sehe, dass mein Gedächtnis fehlbar ist! Ich habe den Namen aktualisiert und einen Link hinzugefügt, um sicherzugehen.
Mdma
Um "für immer" zu warten, verwenden Sie es wie awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS); stackoverflow.com/a/1250655/32453
rogerdpack
Ich denke, dies ist der einfachste Ansatz
Shervin Asgari
1
@ MosheElisha, bist du sicher?. docs.oracle.com/javase/8/docs/api/java/util/concurrent/… sagt Initiiert ein ordnungsgemäßes Herunterfahren, bei dem zuvor übermittelte Aufgaben ausgeführt werden, aber keine neuen Aufgaben akzeptiert werden.
Jaime Hablutzel
7

Sie können warten, bis die Jobs in einem bestimmten Intervall beendet sind:

int maxSecondsPerComputeDTask = 20;
try {
    while (!es.awaitTermination(uniquePhrases.size() * maxSecondsPerComputeDTask, TimeUnit.SECONDS)) {
        // consider giving up with a 'break' statement under certain conditions
    }
} catch (InterruptedException e) {
    throw new RuntimeException(e);    
}

Oder Sie könnten ExecutorService verwenden . einreichen ( Runnable ) und sammeln Sie die Zukunft Objekte , die es gibt und Call get () auf jedem wiederum zu warten , bis sie zu beenden.

ExecutorService es = Executors.newFixedThreadPool(2);
Collection<Future<?>> futures = new LinkedList<<Future<?>>();
for (DataTable singleTable : uniquePhrases) {
    futures.add(es.submit(new ComputeDTask(singleTable)));
}
for (Future<?> future : futures) {
   try {
       future.get();
   } catch (InterruptedException e) {
       throw new RuntimeException(e);
   } catch (ExecutionException e) {
       throw new RuntimeException(e);
   }
}

InterruptedException ist äußerst wichtig, um richtig zu handeln. Auf diese Weise können Sie oder die Benutzer Ihrer Bibliothek einen langen Prozess sicher beenden.

Alain O'Dea
quelle
6

Benutz einfach

latch = new CountDownLatch(noThreads)

In jedem Thread

latch.countDown();

und als Barriere

latch.await();
J. Ruhe
quelle
6

Grundursache für die IllegalMonitorStateException :

Wird ausgelöst, um anzuzeigen, dass ein Thread versucht hat, auf dem Monitor eines Objekts zu warten, oder um andere Threads zu benachrichtigen, die auf dem Monitor eines Objekts warten, ohne den angegebenen Monitor zu besitzen.

Von Ihrem Code aus haben Sie gerade wait () für ExecutorService aufgerufen, ohne die Sperre zu besitzen.

Der folgende Code wird behoben IllegalMonitorStateException

try 
{
    synchronized(es){
        es.wait(); // Add some condition before you call wait()
    }
} 

Befolgen Sie einen der folgenden Ansätze, um auf die Fertigstellung aller Aufgaben zu warten, die an gesendet wurden ExecutorService.

  1. Durchlaufen Sie alle FutureAufgaben von Anfang submitan ExecutorServiceund überprüfen Sie den Status, indem Sie den Aufruf get()des FutureObjekts blockieren

  2. Verwenden von invokeAll onExecutorService

  3. Mit CountDownLatch

  4. Verwenden von ForkJoinPool oder newWorkStealingPool von Executors(seit Java 8)

  5. Shutdown der Pool wie in der Oracle - Dokumentation empfohlen Seite

    void shutdownAndAwaitTermination(ExecutorService pool) {
       pool.shutdown(); // Disable new tasks from being submitted
       try {
       // Wait a while for existing tasks to terminate
       if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
           pool.shutdownNow(); // Cancel currently executing tasks
           // Wait a while for tasks to respond to being cancelled
           if (!pool.awaitTermination(60, TimeUnit.SECONDS))
           System.err.println("Pool did not terminate");
       }
    } catch (InterruptedException ie) {
         // (Re-)Cancel if current thread also interrupted
         pool.shutdownNow();
         // Preserve interrupt status
         Thread.currentThread().interrupt();
    }

    Wenn Sie ordnungsgemäß auf den Abschluss aller Aufgaben warten möchten, wenn Sie Option 5 anstelle der Optionen 1 bis 4 verwenden, ändern Sie diese

    if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {

    zu

    a, while(condition)die alle 1 Minute überprüft.

Ravindra Babu
quelle
6

Sie können die ExecutorService.invokeAllMethode verwenden. Sie führt alle Aufgaben aus und wartet, bis alle Threads ihre Aufgabe beendet haben.

Hier ist komplettes Javadoc

Sie können auch eine überladene Version dieser Methode verwenden, um das Zeitlimit anzugeben.

Hier ist Beispielcode mit ExecutorService.invokeAll

public class Test {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService service = Executors.newFixedThreadPool(3);
        List<Callable<String>> taskList = new ArrayList<>();
        taskList.add(new Task1());
        taskList.add(new Task2());
        List<Future<String>> results = service.invokeAll(taskList);
        for (Future<String> f : results) {
            System.out.println(f.get());
        }
    }

}

class Task1 implements Callable<String> {
    @Override
    public String call() throws Exception {
        try {
            Thread.sleep(2000);
            return "Task 1 done";
        } catch (Exception e) {
            e.printStackTrace();
            return " error in task1";
        }
    }
}

class Task2 implements Callable<String> {
    @Override
    public String call() throws Exception {
        try {
            Thread.sleep(3000);
            return "Task 2 done";
        } catch (Exception e) {
            e.printStackTrace();
            return " error in task2";
        }
    }
}
Nitin Vavdiya
quelle
3

Ich habe auch die Situation, dass ich eine Reihe von Dokumenten zum Crawlen habe. Ich beginne mit einem anfänglichen "Seed" -Dokument, das verarbeitet werden soll. Dieses Dokument enthält Links zu anderen Dokumenten, die ebenfalls verarbeitet werden sollen, und so weiter.

In meinem Hauptprogramm möchte ich nur Folgendes schreiben, in dem Crawlereine Reihe von Threads gesteuert werden.

Crawler c = new Crawler();
c.schedule(seedDocument); 
c.waitUntilCompletion()

Die gleiche Situation würde passieren, wenn ich durch einen Baum navigieren wollte; Ich würde den Stammknoten einfügen, der Prozessor für jeden Knoten würde der Warteschlange nach Bedarf untergeordnete Elemente hinzufügen, und eine Reihe von Threads würde alle Knoten im Baum verarbeiten, bis keine mehr vorhanden wären.

Ich konnte in der JVM nichts finden, was ich etwas überraschend fand. Also habe ich eine Klasse geschrieben, ThreadPooldie man entweder direkt oder in einer Unterklasse verwenden kann, um für die Domäne geeignete Methoden hinzuzufügen, z schedule(Document). Ich hoffe es hilft!

ThreadPool Javadoc | Maven

Adrian Smith
quelle
Doc Link ist tot
Manti_Core
@Manti_Core - danke, aktualisiert.
Adrian Smith
2

Fügen Sie alle Threads in der Sammlung hinzu und senden Sie sie mit invokeAll. Wenn Sie die invokeAllMethode von verwenden können ExecutorService, fährt JVM erst mit der nächsten Zeile fort, wenn alle Threads abgeschlossen sind.

Hier gibt es ein gutes Beispiel: invokeAll via ExecutorService

zgormez
quelle
1

Senden Sie Ihre Aufgaben an den Runner und warten Sie, bis Sie die Methode waitTillDone () wie folgt aufrufen :

Runner runner = Runner.runner(2);

for (DataTable singleTable : uniquePhrases) {

    runner.run(new ComputeDTask(singleTable));
}

// blocks until all tasks are finished (or failed)
runner.waitTillDone();

runner.shutdown();

Um es zu verwenden, fügen Sie diese Gradle / Maven-Abhängigkeit hinzu: 'com.github.matejtymes:javafixes:1.0'

Weitere Informationen finden Sie hier: https://github.com/MatejTymes/JavaFixes oder hier: http://matejtymes.blogspot.com/2016/04/executor-that-notifications-you-when-task.html

Matej Tymes
quelle
0

Eine einfache Alternative dazu ist die Verwendung von Threads zusammen mit Join. Siehe: Threads verbinden

Vicky Kapadia
quelle
3
ExecutorServices macht die Dinge einfacher
David Mann
0

Ich werde nur warten, bis der Executor mit einem festgelegten Timeout beendet ist, das Ihrer Meinung nach für die Ausführung der Aufgaben geeignet ist.

 try {  
         //do stuff here 
         exe.execute(thread);
    } finally {
        exe.shutdown();
    }
    boolean result = exe.awaitTermination(4, TimeUnit.HOURS);
    if (!result)

    {
        LOGGER.error("It took more than 4 hour for the executor to stop, this shouldn't be the normal behaviour.");
    }
Punker
quelle
0

Klingt so, als ob Sie ForkJoinPoolden globalen Pool benötigen und verwenden, um Aufgaben auszuführen.

public static void main(String[] args) {
    // the default `commonPool` should be sufficient for many cases.
    ForkJoinPool pool = ForkJoinPool.commonPool(); 
    // The root of your task that may spawn other tasks. 
    // Make sure it submits the additional tasks to the same executor that it is in.
    Runnable rootTask = new YourTask(pool); 
    pool.execute(rootTask);
    pool.awaitQuiescence(...);
    // that's it.
}

Das Schöne ist, pool.awaitQuiescencedass die Methode den Thread des Aufrufers blockiert, um seine Aufgaben auszuführen, und dann zurückkehrt, wenn er wirklich leer ist.

adib
quelle