Warten Sie, bis alle Threads ihre Arbeit in Java beendet haben

91

Ich schreibe eine Anwendung mit 5 Threads, die gleichzeitig Informationen aus dem Web abrufen und 5 verschiedene Felder in einer Pufferklasse ausfüllen.
Ich muss Pufferdaten validieren und in einer Datenbank speichern, wenn alle Threads ihren Job beendet haben.
Wie kann ich das tun (benachrichtigt werden, wenn alle Threads ihre Arbeit beendet haben)?

RYN
quelle
4
Thread.join ist eine sehr einfache, sehr Java-eigenwillige Methode, um das Problem zu lösen. Darüber hinaus ist dies problematisch, da die Thread- API fehlerhaft ist: Sie können nicht wissen, ob der Join erfolgreich abgeschlossen wurde oder nicht (siehe Java Concurrency In Practice ). Eine Abstraktion auf höherer Ebene, wie die Verwendung eines CountDownLatch, ist möglicherweise vorzuziehen und sieht für Programmierer, die nicht in der Java-idiosynchratischen Denkweise "stecken", natürlicher aus. Streite nicht mit mir, streite mit Doug Lea; )
Cedric Martin

Antworten:

118

Der Ansatz, den ich verfolge, besteht darin, einen ExecutorService zum Verwalten von Thread-Pools zu verwenden.

ExecutorService es = Executors.newCachedThreadPool();
for(int i=0;i<5;i++)
    es.execute(new Runnable() { /*  your task */ });
es.shutdown();
boolean finished = es.awaitTermination(1, TimeUnit.MINUTES);
// all tasks have finished or the time has been reached.
Peter Lawrey
quelle
7
@Leonid genau das macht shutdown ().
Peter Lawrey
3
while(!es.awaitTermination(1, TimeUnit.MINUTES));
Wassermann Power
3
@AquariusPower Sie könnten ihm einfach sagen, dass er länger oder für immer warten soll.
Peter Lawrey
1
Oh ich verstehe; Deshalb habe ich der Schleife eine Nachricht hinzugefügt, die besagt, dass alle Threads auf ihren Abschluss warten. Vielen Dank!
Wassermann Power
1
@ PeterLawrey, ist es notwendig anzurufen es.shutdown();? Was ist, wenn ich einen Code schreibe, in dem ich einen Thread mit es.execute(runnableObj_ZipMaking);in tryblock ausgeführt und in finallyaufgerufen habe boolean finshed = es.awaitTermination(10, TimeUnit.MINUTES);? Ich nehme an, dies sollte warten, bis alle Threads ihre Arbeit abgeschlossen haben oder eine Zeitüberschreitung eingetreten ist (was auch immer zuerst ist). Ist meine Annahme richtig? oder anrufen shutdown()ist obligatorisch?
Amogh
52

Sie können joinzu den Threads. Der Join wird blockiert, bis der Thread abgeschlossen ist.

for (Thread thread : threads) {
    thread.join();
}

Beachten Sie, dass joinein InterruptedException. In diesem Fall müssen Sie entscheiden, was zu tun ist (z. B. versuchen Sie, die anderen Threads abzubrechen, um unnötige Arbeiten zu vermeiden).

Mark Byers
quelle
1
Laufen diese Threads parallel oder nacheinander?
James Webster
5
@ JamesWebster: Parallel.
RYN
4
@ James Webster: Die Anweisung t.join();bedeutet, dass der aktuelle Thread blockiert, bis der Thread tbeendet wird. Der Thread ist davon nicht betroffen t.
Mark Byers
1
Vielen Dank. =] Paralellismus an der Uni studiert, aber das war das einzige, was ich nur schwer lernen konnte! Zum Glück muss ich es jetzt nicht viel verwenden oder wenn ich es tue, ist es nicht zu komplex oder es gibt keine gemeinsam genutzten Ressourcen und das Blockieren ist nicht kritisch
James Webster
1
@ 4r1y4n Ob der bereitgestellte Code wirklich parallel ist, hängt davon ab, was Sie damit tun möchten, und hängt mehr mit der Aggregation von Daten zusammen, die mithilfe verknüpfter Threads über Sammlungen verteilt sind. Sie verbinden Threads, was möglicherweise bedeutet, dass Daten "verbunden" werden. Parallelität bedeutet NICHT unbedingt Parallelität. Das hängt von den CPUs ab. Es kann sehr gut sein, dass die Threads parallel ausgeführt werden, die Berechnungen jedoch in der von der zugrunde liegenden CPU festgelegten Reihenfolge erfolgen.
22

Schauen Sie sich verschiedene Lösungen an.

  1. join()Die API wurde in früheren Versionen von Java eingeführt. Seit der Veröffentlichung von JDK 1.5 sind mit diesem gleichzeitigen Paket einige gute Alternativen verfügbar .

  2. ExecutorService # invokeAll ()

    Führt die angegebenen Aufgaben aus und gibt eine Liste der Futures zurück, die ihren Status und ihre Ergebnisse enthalten, wenn alles abgeschlossen ist.

    In dieser verwandten SE-Frage finden Sie ein Codebeispiel:

    Wie verwende ich invokeAll (), um alle Thread-Pools ihre Aufgabe erledigen zu lassen?

  3. CountDownLatch

    Eine Synchronisationshilfe, mit der ein oder mehrere Threads warten können, bis eine Reihe von Vorgängen in anderen Threads abgeschlossen ist.

    Ein CountDownLatch wird mit einer bestimmten Anzahl initialisiert. Die Wartemethoden blockieren, bis der aktuelle Zählerstand aufgrund von Aufrufen der countDown()Methode Null erreicht. Danach werden alle wartenden Threads freigegeben und alle nachfolgenden Aufrufe von Wartezeiten werden sofort zurückgegeben. Dies ist ein One-Shot-Phänomen - die Zählung kann nicht zurückgesetzt werden. Wenn Sie eine Version benötigen, die die Anzahl zurücksetzt, sollten Sie einen CyclicBarrier verwenden .

    Beziehen Sie sich auf diese Frage für die Verwendung von CountDownLatch

    Wie kann man auf einen Thread warten, der einen eigenen Thread erzeugt?

  4. ForkJoinPool oder newWorkStealingPool () in Executors

  5. Durchlaufen Sie alle zukünftigen Objekte, die nach dem Senden an erstellt wurdenExecutorService

Ravindra Babu
quelle
11

Neben den Thread.join()Vorschlägen anderer führte Java 5 das Executor-Framework ein. Dort arbeitet man nicht mit ThreadObjekten. Stattdessen senden Sie Ihre Callableoder RunnableObjekte an einen Executor. Es gibt einen speziellen Executor, der mehrere Aufgaben ausführen und deren Ergebnisse nicht in der richtigen Reihenfolge zurückgeben soll. Das ist das ExecutorCompletionService:

ExecutorCompletionService executor;
for (..) {
    executor.submit(Executors.callable(yourRunnable));
}

Dann können Sie wiederholt aufrufen, take()bis keine Future<?>Objekte mehr zurückgegeben werden müssen, was bedeutet, dass alle abgeschlossen sind.


Eine andere Sache, die abhängig von Ihrem Szenario relevant sein kann, ist CyclicBarrier.

Eine Synchronisationshilfe, mit der eine Reihe von Threads darauf warten können, dass sie einen gemeinsamen Barrierepunkt erreichen. CyclicBarriers sind nützlich in Programmen, an denen eine Gruppe von Threads fester Größe beteiligt ist, die gelegentlich aufeinander warten müssen. Die Barriere wird als zyklisch bezeichnet, da sie nach dem Lösen der wartenden Threads wiederverwendet werden kann.

Bozho
quelle
Dies ist nah, aber ich würde noch ein paar Anpassungen vornehmen. executor.submitgibt a zurück Future<?>. Ich würde diese Futures zu einer Liste hinzufügen und dann die Liste durchlaufen, die getjede Zukunft aufruft .
Ray
Sie können einen Konstruktor auch instanziieren, indem Sie Executorsz. B. Executors.newCachedThreadPool(oder ähnliches)
Ray
10

Eine andere Möglichkeit ist das CountDownLatchObjekt, das für einfache Situationen nützlich ist: Da Sie die Anzahl der Threads im Voraus kennen, initialisieren Sie es mit der entsprechenden Anzahl und übergeben die Referenz des Objekts an jeden Thread.
Nach Abschluss seiner Aufgabe ruft jeder Thread auf, CountDownLatch.countDown()wodurch der interne Zähler dekrementiert wird. Der Hauptthread sollte nach dem Starten aller anderen den CountDownLatch.await()Blockierungsaufruf ausführen. Es wird freigegeben, sobald der interne Zähler 0 erreicht hat.

Achten Sie darauf, dass mit diesem Objekt auch ein InterruptedExceptiongeworfen werden kann.

interDist
quelle
8

Sie machen

for (Thread t : new Thread[] { th1, th2, th3, th4, th5 })
    t.join()

Nach dieser for-Schleife können Sie sicher sein, dass alle Threads ihre Jobs beendet haben.

aioobe
quelle
5

Warten / blockieren Sie den Thread Main, bis einige andere Threads ihre Arbeit abgeschlossen haben.

Wie @Ravindra babugesagt, es kann auf verschiedene Arten erreicht werden, aber anhand von Beispielen gezeigt werden.

  • java.lang.Thread. join () Seit: 1.0

    public static void joiningThreads() throws InterruptedException {
        Thread t1 = new Thread( new LatchTask(1, null), "T1" );
        Thread t2 = new Thread( new LatchTask(7, null), "T2" );
        Thread t3 = new Thread( new LatchTask(5, null), "T3" );
        Thread t4 = new Thread( new LatchTask(2, null), "T4" );
    
        // Start all the threads
        t1.start();
        t2.start();
        t3.start();
        t4.start();
    
        // Wait till all threads completes
        t1.join();
        t2.join();
        t3.join();
        t4.join();
    }
  • java.util.concurrent.CountDownLatch Seit: 1.5

    • .countDown() «Verringert die Anzahl der Latch-Gruppen.
    • .await() «Die Wartemethoden blockieren, bis die aktuelle Anzahl Null erreicht.

    Wenn Sie erstellt latchGroupCount = 4dann countDown()sollten 4 mal aufgerufen werden , um 0. So zu stellen rechnen, dass await()wird die Sperrfäden lösen.

    public static void latchThreads() throws InterruptedException {
        int latchGroupCount = 4;
        CountDownLatch latch = new CountDownLatch(latchGroupCount);
        Thread t1 = new Thread( new LatchTask(1, latch), "T1" );
        Thread t2 = new Thread( new LatchTask(7, latch), "T2" );
        Thread t3 = new Thread( new LatchTask(5, latch), "T3" );
        Thread t4 = new Thread( new LatchTask(2, latch), "T4" );
    
        t1.start();
        t2.start();
        t3.start();
        t4.start();
    
        //latch.countDown();
    
        latch.await(); // block until latchGroupCount is 0.
    }

Beispielcode der Threaded-Klasse LatchTask. Um den Ansatz zu testen, verwenden Sie joiningThreads(); und latchThreads();von der Hauptmethode.

class LatchTask extends Thread {
    CountDownLatch latch;
    int iterations = 10;
    public LatchTask(int iterations, CountDownLatch latch) {
        this.iterations = iterations;
        this.latch = latch;
    }

    @Override
    public void run() {
        String threadName = Thread.currentThread().getName();
        System.out.println(threadName + " : Started Task...");

        for (int i = 0; i < iterations; i++) {
            System.out.println(threadName + " : " + i);
            MainThread_Wait_TillWorkerThreadsComplete.sleep(1);
        }
        System.out.println(threadName + " : Completed Task");
        // countDown() « Decrements the count of the latch group.
        if(latch != null)
            latch.countDown();
    }
}
  • CyclicBarriers Eine Synchronisationshilfe, mit der eine Reihe von Threads darauf warten können, dass alle einen gemeinsamen Barrierepunkt erreichen. CyclicBarriers sind in Programmen nützlich, an denen eine Gruppe von Threads fester Größe beteiligt ist, die gelegentlich aufeinander warten müssen. Die Barriere wird als zyklisch bezeichnet, da sie nach dem Lösen der wartenden Threads wiederverwendet werden kann.
    CyclicBarrier barrier = new CyclicBarrier(3);
    barrier.await();
    Verweisen Sie beispielsweise auf diese Concurrent_ParallelNotifyies- Klasse.

  • Executer-Framework: Mit ExecutorService können wir einen Thread-Pool erstellen und den Fortschritt der asynchronen Aufgaben mit Future verfolgen.

    • submit(Runnable), submit(Callable)die Future Object zurückgeben. Mit der future.get()Funktion können wir den Haupt-Thread blockieren, bis der Arbeitsthread seine Arbeit beendet hat.

    • invokeAll(...) - gibt eine Liste zukünftiger Objekte zurück, über die Sie die Ergebnisse der Ausführungen der einzelnen Callable abrufen können.

Hier finden Sie ein Beispiel für die Verwendung von Interfaces Runnable, Callable with Executor Framework.


@Siehe auch

Yash
quelle
4

Speichern Sie die Thread-Objekte in einer Sammlung (wie einer Liste oder einem Set), durchlaufen Sie dann die Sammlung, sobald die Threads gestartet sind, und rufen Sie join () für die Threads auf.

esaj
quelle
2

Obwohl dies für das Problem von OP nicht relevant ist, können Sie einen Exchanger verwenden , wenn Sie an einer Synchronisation (genauer gesagt einem Rendez-Vous) mit genau einem Thread interessiert sind

In meinem Fall musste ich den übergeordneten Thread anhalten, bis der untergeordnete Thread etwas getan hat, z. B. seine Initialisierung abgeschlossen hat. Ein CountDownLatch funktioniert auch gut.

18446744073709551615
quelle
1

versuchen Sie dies, wird funktionieren.

  Thread[] threads = new Thread[10];

  List<Thread> allThreads = new ArrayList<Thread>();

  for(Thread thread : threads){

        if(null != thread){

              if(thread.isAlive()){

                    allThreads.add(thread);

              }

        }

  }

  while(!allThreads.isEmpty()){

        Iterator<Thread> ite = allThreads.iterator();

        while(ite.hasNext()){

              Thread thread = ite.next();

              if(!thread.isAlive()){

                   ite.remove();
              }

        }

   }
Jeyaraj.J
quelle
1

Ich hatte ein ähnliches Problem und verwendete schließlich Java 8 parallelStream.

requestList.parallelStream().forEach(req -> makeRequest(req));

Es ist super einfach und lesbar. Hinter den Kulissen wird der Standard-Fork-Join-Pool von JVM verwendet. Dies bedeutet, dass gewartet wird, bis alle Threads abgeschlossen sind, bevor Sie fortfahren. Für meinen Fall war es eine ordentliche Lösung, da es der einzige parallelStream in meiner Anwendung war. Wenn mehr als ein parallelStream gleichzeitig ausgeführt wird, lesen Sie bitte den folgenden Link.

Weitere Informationen zu parallelen Streams finden Sie hier .

Madis Pukkonen
quelle
0

Die vorhandenen Antworten besagten, könnte join()jeder Thread.

Es gibt jedoch mehrere Möglichkeiten, um das Thread-Array / die Thread-Liste abzurufen:

  • Fügen Sie den Thread bei der Erstellung einer Liste hinzu.
  • Verwenden Sie ThreadGroupdiese Option , um die Threads zu verwalten.

Der folgende Code verwendet den ThreadGruopAnsatz. Es wird zuerst eine Gruppe erstellt. Wenn Sie dann jeden Thread erstellen, geben Sie die Gruppe im Konstruktor an. Später kann das Thread-Array über abgerufen werdenThreadGroup.enumerate()


Code

SyncBlockLearn.java

import org.testng.Assert;
import org.testng.annotations.Test;

/**
 * synchronized block - learn,
 *
 * @author eric
 * @date Apr 20, 2015 1:37:11 PM
 */
public class SyncBlockLearn {
    private static final int TD_COUNT = 5; // thread count
    private static final int ROUND_PER_THREAD = 100; // round for each thread,
    private static final long INC_DELAY = 10; // delay of each increase,

    // sync block test,
    @Test
    public void syncBlockTest() throws InterruptedException {
        Counter ct = new Counter();
        ThreadGroup tg = new ThreadGroup("runner");

        for (int i = 0; i < TD_COUNT; i++) {
            new Thread(tg, ct, "t-" + i).start();
        }

        Thread[] tArr = new Thread[TD_COUNT];
        tg.enumerate(tArr); // get threads,

        // wait all runner to finish,
        for (Thread t : tArr) {
            t.join();
        }

        System.out.printf("\nfinal count: %d\n", ct.getCount());
        Assert.assertEquals(ct.getCount(), TD_COUNT * ROUND_PER_THREAD);
    }

    static class Counter implements Runnable {
        private final Object lkOn = new Object(); // the object to lock on,
        private int count = 0;

        @Override
        public void run() {
            System.out.printf("[%s] begin\n", Thread.currentThread().getName());

            for (int i = 0; i < ROUND_PER_THREAD; i++) {
                synchronized (lkOn) {
                    System.out.printf("[%s] [%d] inc to: %d\n", Thread.currentThread().getName(), i, ++count);
                }
                try {
                    Thread.sleep(INC_DELAY); // wait a while,
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            System.out.printf("[%s] end\n", Thread.currentThread().getName());
        }

        public int getCount() {
            return count;
        }
    }
}

Der Haupt-Thread wartet, bis alle Threads in der Gruppe beendet sind.

Eric Wang
quelle
0

Ich habe eine kleine Hilfsmethode erstellt, um auf das Ende einiger Threads zu warten:

public static void waitForThreadsToFinish(Thread... threads) {
        try {
            for (Thread thread : threads) {
                thread.join();
            }
        }
        catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
Alon Gouldman
quelle
-1

Verwenden Sie dies in Ihrem Hauptthread: while (! Executor.isTerminated ()); Fügen Sie diese Codezeile ein, nachdem Sie alle Threads vom Executor-Dienst gestartet haben. Dadurch wird der Hauptthread erst gestartet, nachdem alle von Executoren gestarteten Threads abgeschlossen sind. Stellen Sie sicher, dass Sie executor.shutdown () aufrufen. vor der obigen Schleife.

Maggy
quelle
Dies ist ein aktives Warten, wodurch die CPU ständig eine leere Schleife ausführt. Sehr verschwenderisch.
Adam Michalik