Wie kann ich warten, bis einige Threads abgeschlossen sind?

109

Wie kann man einfach warten, bis der gesamte Thread-Prozess abgeschlossen ist? Nehmen wir zum Beispiel an, ich habe:

public class DoSomethingInAThread implements Runnable{

    public static void main(String[] args) {
        for (int n=0; n<1000; n++) {
            Thread t = new Thread(new DoSomethingInAThread());
            t.start();
        }
        // wait for all threads' run() methods to complete before continuing
    }

    public void run() {
        // do something here
    }


}

Wie ändere ich dies, damit die main()Methode beim Kommentar pausiert, bis alle run()Methoden der Threads beendet werden? Vielen Dank!

DivideByHero
quelle

Antworten:

163

Sie fügen alle Threads in ein Array ein, starten sie alle und haben dann eine Schleife

for(i = 0; i < threads.length; i++)
  threads[i].join();

Jeder Join wird blockiert, bis der jeweilige Thread abgeschlossen ist. Threads werden möglicherweise in einer anderen Reihenfolge abgeschlossen als Sie, aber das ist kein Problem: Wenn die Schleife beendet wird, sind alle Threads abgeschlossen.

Martin v. Löwis
quelle
1
@ Mykola: Was genau ist der Vorteil einer Thread-Gruppe? Nur weil die API vorhanden ist, heißt das nicht, dass Sie sie verwenden müssen ...
Martin v. Löwis
2
Siehe: "Eine Thread-Gruppe repräsentiert eine Reihe von Threads." Dies ist für diesen Anwendungsfall semantisch korrekt! Und: "Ein Thread darf auf Informationen über seine eigene Thread-Gruppe zugreifen"
Martin K.
4
Das Buch „Effective Java“ empfiehlt, Thread-Gruppen zu vermeiden (Punkt 73).
Bastien Léonard
2
Die in Effective Java erwähnten Fehler sollten in Java 6 behoben worden sein. Wenn neuere Java-Versionen keine Einschränkung darstellen, ist es besser, Futures zu verwenden, um Thread-Probleme zu lösen. Martin v. Löwis: Sie haben recht. Es ist nicht relevant für dieses Problem, aber es ist schön, mehr Informationen über die laufenden Threads von einem Objekt (wie dem ExecutorService) zu erhalten. Ich finde es schön, bestimmte Funktionen zu verwenden, um ein Problem zu lösen. Vielleicht benötigen Sie in Zukunft mehr Flexibilität (Thread-Informationen). Es ist auch richtig, die alten Buggy-Klassen in älteren JDKs zu erwähnen.
Martin K.
5
ThreadGroup implementiert keinen Join auf Gruppenebene. Warum die Leute ThreadGroup pushen, ist also ein wenig verwirrend. Verwenden die Leute wirklich Drehsperren und fragen sie den activeCount der Gruppe ab? Es fällt Ihnen schwer, mich davon zu überzeugen, dass dies in irgendeiner Weise besser ist, als wenn Sie nur Join für alle Threads aufrufen.
41

Eine Möglichkeit wäre, eine machen Listvon Threads, erstellen und jeden Thread zu starten, während es in die Liste. Wenn alles gestartet ist, durchlaufen Sie die Liste erneut und rufen Sie join()jede einzelne auf. Es spielt keine Rolle, in welcher Reihenfolge die Threads ausgeführt werden. Sie müssen lediglich wissen, dass bis zum Ende der Ausführung der zweiten Schleife jeder Thread abgeschlossen ist.

Ein besserer Ansatz ist die Verwendung eines ExecutorService und der zugehörigen Methoden:

List<Callable> callables = ... // assemble list of Callables here
                               // Like Runnable but can return a value
ExecutorService execSvc = Executors.newCachedThreadPool();
List<Future<?>> results = execSvc.invokeAll(callables);
// Note: You may not care about the return values, in which case don't
//       bother saving them

Die Verwendung eines ExecutorService (und aller neuen Funktionen der Parallelitätsdienstprogramme von Java 5 ) ist unglaublich flexibel, und das obige Beispiel kratzt kaum an der Oberfläche.

Adam Batkin
quelle
ThreadGroup ist der richtige Weg! Mit einer veränderlichen Liste geraten Sie in Schwierigkeiten (Synchronisation)
Martin K.
3
Was? Wie würden Sie in Schwierigkeiten geraten? Es ist nur für den Thread, der den Start ausführt, veränderlich (nur lesbar). Solange die Liste während des Durchlaufens nicht geändert wird, ist dies in Ordnung.
Adam Batkin
Es hängt davon ab, wie Sie es verwenden. Wenn Sie die aufrufende Klasse in einem Thread verwenden, treten Probleme auf.
Martin K.
27
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class DoSomethingInAThread implements Runnable
{
   public static void main(String[] args) throws ExecutionException, InterruptedException
   {
      //limit the number of actual threads
      int poolSize = 10;
      ExecutorService service = Executors.newFixedThreadPool(poolSize);
      List<Future<Runnable>> futures = new ArrayList<Future<Runnable>>();

      for (int n = 0; n < 1000; n++)
      {
         Future f = service.submit(new DoSomethingInAThread());
         futures.add(f);
      }

      // wait for all tasks to complete before continuing
      for (Future<Runnable> f : futures)
      {
         f.get();
      }

      //shut down the executor service so that this thread can exit
      service.shutdownNow();
   }

   public void run()
   {
      // do something here
   }
}
jt.
quelle
hat wie ein Zauber funktioniert ... Ich habe zwei Sätze von Threads, die aufgrund von Problemen mit mehreren Cookies nicht gleichzeitig ausgeführt werden sollten. Ich habe Ihr Beispiel verwendet, um jeweils einen Satz von Threads
auszuführen. Vielen
@Dantalian - In Ihrer Runnable-Klasse (wahrscheinlich in der Ausführungsmethode) möchten Sie alle aufgetretenen Ausnahmen erfassen und lokal speichern (oder eine Fehlermeldung / Bedingung speichern). Im Beispiel gibt f.get () Ihr Objekt zurück, das Sie an den ExecutorService gesendet haben. Ihr Objekt verfügt möglicherweise über eine Methode zum Abrufen von Ausnahmen / Fehlerbedingungen. Abhängig davon, wie Sie das bereitgestellte Beispiel ändern, müssen Sie möglicherweise das von f.get () gedrehte Objekt in den erwarteten Typ umwandeln.
jt.
12

Anstelle join()einer alten API können Sie auch CountDownLatch verwenden . Ich habe Ihren Code wie folgt geändert, um Ihre Anforderung zu erfüllen.

import java.util.concurrent.*;
class DoSomethingInAThread implements Runnable{
    CountDownLatch latch;
    public DoSomethingInAThread(CountDownLatch latch){
        this.latch = latch;
    } 
    public void run() {
        try{
            System.out.println("Do some thing");
            latch.countDown();
        }catch(Exception err){
            err.printStackTrace();
        }
    }
}

public class CountDownLatchDemo {
    public static void main(String[] args) {
        try{
            CountDownLatch latch = new CountDownLatch(1000);
            for (int n=0; n<1000; n++) {
                Thread t = new Thread(new DoSomethingInAThread(latch));
                t.start();
            }
            latch.await();
            System.out.println("In Main thread after completion of 1000 threads");
        }catch(Exception err){
            err.printStackTrace();
        }
    }
}

Erklärung :

  1. CountDownLatch wurde mit der angegebenen Anzahl 1000 gemäß Ihrer Anforderung initialisiert.

  2. Jeder Worker-Thread DoSomethingInAThread dekrementiert den CountDownLatchim Konstruktor übergebenen Thread .

  3. Hauptfaden CountDownLatchDemo await()bis die Zählung Null geworden ist. Sobald der Zähler Null geworden ist, werden Sie in der Ausgabe unter die Linie gebracht.

    In Main thread after completion of 1000 threads

Weitere Informationen finden Sie auf der Oracle-Dokumentationsseite

public void await()
           throws InterruptedException

Bewirkt, dass der aktuelle Thread wartet, bis der Latch auf Null heruntergezählt wurde, es sei denn, der Thread wird unterbrochen.

Weitere Optionen finden Sie in der entsprechenden SE-Frage:

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

Ravindra Babu
quelle
8

Vermeiden Sie die Thread-Klasse insgesamt und verwenden Sie stattdessen die höheren Abstraktionen in java.util.concurrent

Die ExecutorService-Klasse stellt die Methode invokeAll bereit , die genau das zu tun scheint, was Sie wollen.

Henrik
quelle
6

Erwägen Sie die Verwendung java.util.concurrent.CountDownLatch. Beispiele in Javadocs

Pablo Cavalieri
quelle
Ist ein Latch für Threads, arbeitet die Latch-Sperre mit einem Countdown. In der run () -Methode Ihres Threads wird explizit deklariert, dass darauf gewartet werden soll, dass ein CountDownLatch seinen Countdown bis 0 erreicht. Sie können denselben CountDownLatch in mehr als einem Thread verwenden, um sie gleichzeitig freizugeben. Ich weiß nicht, ob es das ist, was Sie brauchen, wollte es nur erwähnen, weil es nützlich ist, wenn Sie in einer Multithread-Umgebung arbeiten.
Pablo Cavalieri
Vielleicht sollten Sie diese Erklärung in Ihre Antwort einfügen?
Aaron Hall
Die Beispiele im Javadoc sind sehr beschreibend, deshalb habe ich keine hinzugefügt. docs.oracle.com/javase/7/docs/api/java/util/concurrent/… . Im ersten Beispiel werden alle Worker-Threads gleichzeitig freigegeben, da sie darauf warten, dass das CountdownLatch-Startsignal Null erreicht, was in startSignal.countDown () geschieht. Anschließend wartet der Mian-Thread mit der Anweisung doneSignal.await (), bis alle Arbeiten abgeschlossen sind. doneSignal verringert den Wert in jedem Worker.
Pablo Cavalieri
6

Wie Martin K vorgeschlagen hat, java.util.concurrent.CountDownLatchscheint dies eine bessere Lösung zu sein. Fügen Sie einfach ein Beispiel dafür hinzu

     public class CountDownLatchDemo
{

    public static void main (String[] args)
    {
        int noOfThreads = 5;
        // Declare the count down latch based on the number of threads you need
        // to wait on
        final CountDownLatch executionCompleted = new CountDownLatch(noOfThreads);
        for (int i = 0; i < noOfThreads; i++)
        {
            new Thread()
            {

                @Override
                public void run ()
                {

                    System.out.println("I am executed by :" + Thread.currentThread().getName());
                    try
                    {
                        // Dummy sleep
                        Thread.sleep(3000);
                        // One thread has completed its job
                        executionCompleted.countDown();
                    }
                    catch (InterruptedException e)
                    {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }

            }.start();
        }

        try
        {
            // Wait till the count down latch opens.In the given case till five
            // times countDown method is invoked
            executionCompleted.await();
            System.out.println("All over");
        }
        catch (InterruptedException e)
        {
            e.printStackTrace();
        }
    }

}
Freaky Thommi
quelle
4

Abhängig von Ihren Anforderungen möchten Sie möglicherweise auch die Klassen CountDownLatch und CyclicBarrier im Paket java.util.concurrent überprüfen. Sie können nützlich sein, wenn Sie möchten, dass Ihre Threads aufeinander warten, oder wenn Sie eine genauere Kontrolle über die Ausführung Ihrer Threads wünschen (z. B. wenn Sie in ihrer internen Ausführung darauf warten, dass ein anderer Thread einen bestimmten Status festlegt). Sie können auch einen CountDownLatch verwenden, um zu signalisieren, dass alle Ihre Threads gleichzeitig gestartet werden, anstatt sie einzeln zu starten, während Sie Ihre Schleife durchlaufen. Die Standard-API-Dokumente enthalten ein Beispiel dafür und verwenden einen anderen CountDownLatch, um darauf zu warten, dass alle Threads ihre Ausführung abgeschlossen haben.

Jeff
quelle
1

Erstellen Sie das Thread-Objekt in der ersten for-Schleife.

for (int i = 0; i < threads.length; i++) {
     threads[i] = new Thread(new Runnable() {
         public void run() {
             // some code to run in parallel
         }
     });
     threads[i].start();
 }

Und dann, was alle hier sagen.

for(i = 0; i < threads.length; i++)
  threads[i].join();
optimus0127
quelle
0

Sie können dies mit dem Objekt "ThreadGroup" und seinem Parameter activeCount tun :

Martin K.
quelle
Ich bin mir nicht sicher, wie Sie es genau vorschlagen. Wenn Sie vorschlagen, activeCount in einer Schleife abzufragen, ist dies schlecht, da es mit Wartezeiten beschäftigt ist (selbst wenn Sie zwischen den Umfragen schlafen - Sie erhalten dann einen Kompromiss zwischen Geschäft und Reaktionsfähigkeit).
Martin v. Löwis
@Martin v. Löwis: "Join wartet nur auf einen einzelnen Thread. Eine bessere Lösung könnte ein java.util.concurrent.CountDownLatch sein. Initialisieren Sie den Latch einfach mit der Anzahl, die auf die Anzahl der Worker-Threads festgelegt ist. Jeder Worker-Thread sollte aufrufen countDown () kurz vor dem Beenden und der Hauptthread ruft einfach await () auf, das blockiert, bis der Zähler Null erreicht. Das Problem mit join () ist auch, dass Sie nicht mehr dynamisch weitere Threads hinzufügen können. Die Liste wird explodieren mit einer gleichzeitigen Änderung. " Ihre Lösung funktioniert gut für das Problem, aber nicht für allgemeine Zwecke.
Martin K.
0

Alternativ zu CountDownLatch können Sie auch CyclicBarrier verwenden, z

public class ThreadWaitEx {
    static CyclicBarrier barrier = new CyclicBarrier(100, new Runnable(){
        public void run(){
            System.out.println("clean up job after all tasks are done.");
        }
    });
    public static void main(String[] args) {
        for (int i = 0; i < 100; i++) {
            Thread t = new Thread(new MyCallable(barrier));
            t.start();
        }       
    }

}    

class MyCallable implements Runnable{
    private CyclicBarrier b = null;
    public MyCallable(CyclicBarrier b){
        this.b = b;
    }
    @Override
    public void run(){
        try {
            //do something
            System.out.println(Thread.currentThread().getName()+" is waiting for barrier after completing his job.");
            b.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
    }       
}

Um CyclicBarrier in diesem Fall zu verwenden, sollte barriere.await () die letzte Anweisung sein, dh wenn Ihr Thread mit seiner Arbeit fertig ist. CyclicBarrier kann mit seiner reset () -Methode wieder verwendet werden. Um Javadocs zu zitieren:

Ein CyclicBarrier unterstützt einen optionalen Runnable-Befehl, der einmal pro Barrierepunkt ausgeführt wird, nachdem der letzte Thread in der Gruppe eingetroffen ist, bevor jedoch Threads freigegeben werden. Diese Barriereaktion ist nützlich, um den Status des gemeinsam genutzten Status zu aktualisieren, bevor eine der Parteien fortfährt.

shailendra1118
quelle
Ich denke nicht, dass dies ein gutes Beispiel für CyclicBarrier ist. Warum verwenden Sie einen Thread.sleep () -Aufruf?
Günther
@Guenther - ja, ich habe den Code geändert, um der Anforderung zu entsprechen.
Shailendra1118
CyclicBarrier ist keine Alternative zu CountDownLatch. Wenn Threads wiederholt heruntergezählt werden müssen, sollten Sie einen CyclicBarrier erstellen, andernfalls standardmäßig CountDownLatch (sofern nicht anders eine zusätzliche Abstraktion der Ausführung erforderlich ist. An diesem Punkt sollten Sie sich die übergeordneten Dienste ansehen).
Elysiumplain
0

Das join()war mir nicht hilfreich. siehe dieses Beispiel in Kotlin:

    val timeInMillis = System.currentTimeMillis()
    ThreadUtils.startNewThread(Runnable {
        for (i in 1..5) {
            val t = Thread(Runnable {
                Thread.sleep(50)
                var a = i
                kotlin.io.println(Thread.currentThread().name + "|" + "a=$a")
                Thread.sleep(200)
                for (j in 1..5) {
                    a *= j
                    Thread.sleep(100)
                    kotlin.io.println(Thread.currentThread().name + "|" + "$a*$j=$a")
                }
                kotlin.io.println(Thread.currentThread().name + "|TaskDurationInMillis = " + (System.currentTimeMillis() - timeInMillis))
            })
            t.start()
        }
    })

Das Ergebnis:

Thread-5|a=5
Thread-1|a=1
Thread-3|a=3
Thread-2|a=2
Thread-4|a=4
Thread-2|2*1=2
Thread-3|3*1=3
Thread-1|1*1=1
Thread-5|5*1=5
Thread-4|4*1=4
Thread-1|2*2=2
Thread-5|10*2=10
Thread-3|6*2=6
Thread-4|8*2=8
Thread-2|4*2=4
Thread-3|18*3=18
Thread-1|6*3=6
Thread-5|30*3=30
Thread-2|12*3=12
Thread-4|24*3=24
Thread-4|96*4=96
Thread-2|48*4=48
Thread-5|120*4=120
Thread-1|24*4=24
Thread-3|72*4=72
Thread-5|600*5=600
Thread-4|480*5=480
Thread-3|360*5=360
Thread-1|120*5=120
Thread-2|240*5=240
Thread-1|TaskDurationInMillis = 765
Thread-3|TaskDurationInMillis = 765
Thread-4|TaskDurationInMillis = 765
Thread-5|TaskDurationInMillis = 765
Thread-2|TaskDurationInMillis = 765

Lassen Sie mich jetzt die join()für Threads verwenden:

    val timeInMillis = System.currentTimeMillis()
    ThreadUtils.startNewThread(Runnable {
        for (i in 1..5) {
            val t = Thread(Runnable {
                Thread.sleep(50)
                var a = i
                kotlin.io.println(Thread.currentThread().name + "|" + "a=$a")
                Thread.sleep(200)
                for (j in 1..5) {
                    a *= j
                    Thread.sleep(100)
                    kotlin.io.println(Thread.currentThread().name + "|" + "$a*$j=$a")
                }
                kotlin.io.println(Thread.currentThread().name + "|TaskDurationInMillis = " + (System.currentTimeMillis() - timeInMillis))
            })
            t.start()
            t.join()
        }
    })

Und das Ergebnis:

Thread-1|a=1
Thread-1|1*1=1
Thread-1|2*2=2
Thread-1|6*3=6
Thread-1|24*4=24
Thread-1|120*5=120
Thread-1|TaskDurationInMillis = 815
Thread-2|a=2
Thread-2|2*1=2
Thread-2|4*2=4
Thread-2|12*3=12
Thread-2|48*4=48
Thread-2|240*5=240
Thread-2|TaskDurationInMillis = 1568
Thread-3|a=3
Thread-3|3*1=3
Thread-3|6*2=6
Thread-3|18*3=18
Thread-3|72*4=72
Thread-3|360*5=360
Thread-3|TaskDurationInMillis = 2323
Thread-4|a=4
Thread-4|4*1=4
Thread-4|8*2=8
Thread-4|24*3=24
Thread-4|96*4=96
Thread-4|480*5=480
Thread-4|TaskDurationInMillis = 3078
Thread-5|a=5
Thread-5|5*1=5
Thread-5|10*2=10
Thread-5|30*3=30
Thread-5|120*4=120
Thread-5|600*5=600
Thread-5|TaskDurationInMillis = 3833

Wie es klar ist, wenn wir Folgendes verwenden join:

  1. Die Threads werden nacheinander ausgeführt.
  2. Die erste Stichprobe dauert 765 Millisekunden, während die zweite Stichprobe 3833 Millisekunden dauert.

Unsere Lösung, um das Blockieren anderer Threads zu verhindern, bestand darin, eine ArrayList zu erstellen:

val threads = ArrayList<Thread>()

Wenn wir nun einen neuen Thread starten möchten, fügen wir ihn meistens der ArrayList hinzu:

addThreadToArray(
    ThreadUtils.startNewThread(Runnable {
        ...
    })
)

Die addThreadToArrayFunktion:

@Synchronized
fun addThreadToArray(th: Thread) {
    threads.add(th)
}

Die startNewThreadFunktion:

fun startNewThread(runnable: Runnable) : Thread {
    val th = Thread(runnable)
    th.isDaemon = false
    th.priority = Thread.MAX_PRIORITY
    th.start()
    return th
}

Überprüfen Sie die Vervollständigung der Threads wie unten beschrieben, wo immer dies erforderlich ist:

val notAliveThreads = ArrayList<Thread>()
for (t in threads)
    if (!t.isAlive)
        notAliveThreads.add(t)
threads.removeAll(notAliveThreads)
if (threads.size == 0){
    // The size is 0 -> there is no alive threads.
}
Misagh
quelle