tunggu sampai semua thread menyelesaikan pekerjaannya di java

Saya sedang menulis aplikasi yang memiliki 5 utas yang mendapatkan beberapa informasi dari web secara bersamaan dan mengisi 5 bidang berbeda dalam kelas penyangga.
Saya perlu memvalidasi data buffer dan menyimpannya dalam database ketika semua utas menyelesaikan pekerjaan mereka.
Bagaimana saya bisa melakukan ini (mendapat peringatan ketika semua utas menyelesaikan pekerjaannya)?

Pendekatan yang saya lakukan adalah menggunakan ExecutorService untuk mengelola kumpulan utas.

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.

Anda bisa joinke utas. Blok bergabung sampai utas selesai.

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

Perhatikan bahwa joinmelempar InterruptedException. Anda harus memutuskan apa yang harus dilakukan jika itu terjadi (misalnya mencoba membatalkan utas lain untuk mencegah pekerjaan yang tidak perlu dilakukan).

Lihat berbagai solusi.

1. join()API telah diperkenalkan di versi awal Java. Beberapa alternatif bagus tersedia dengan paket bersamaan ini sejak rilis JDK 1.5.

2. ExecutorService # invokeAll ()

   Melaksanakan tugas yang diberikan, mengembalikan daftar Futures yang memegang status dan hasilnya ketika semuanya selesai.

   Lihat pertanyaan SE terkait ini untuk contoh kode:

   Bagaimana cara menggunakan invokeAll () untuk membiarkan semua kumpulan utas melakukan tugasnya?

3. CountDownLatch

   Bantuan sinkronisasi yang memungkinkan satu atau lebih utas menunggu hingga serangkaian operasi dilakukan di utas lain selesai.

   Sebuah CountDownLatch diinisialisasi dengan jumlah yang diberikan. Metode menunggu memblokir hingga jumlah saat ini mencapai nol karena pemanggilan countDown()metode, setelah itu semua utas menunggu dilepaskan dan pemanggilan await berikutnya segera dikembalikan. Ini adalah fenomena satu bidikan - hitungan tidak dapat disetel ulang. Jika Anda memerlukan versi yang menyetel ulang hitungan, pertimbangkan untuk menggunakan CyclicBarrier .

   Lihat pertanyaan ini untuk penggunaan CountDownLatch

   Bagaimana cara menunggu utas yang memunculkan utasnya sendiri?

4. ForkJoinPool atau newWorkStealingPool () di Pelaku

5. Iterasi melalui semua objek Future yang dibuat setelah mengirimkan keExecutorService

Selain Thread.join()disarankan oleh orang lain, java 5 memperkenalkan kerangka kerja pelaksana. Di sana Anda tidak bekerja dengan Threadobjek. Alih-alih, Anda mengirimkan objek Callableatau Runnableke pelaksana. Ada pelaksana khusus yang dimaksudkan untuk menjalankan banyak tugas dan mengembalikan hasilnya ke dalam urutan. Itu dia ExecutorCompletionService:

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

Kemudian Anda dapat memanggil berulang kali take()hingga tidak ada lagi Future<?>objek untuk dikembalikan, yang berarti semuanya selesai.

Hal lain yang mungkin relevan, tergantung pada skenario Anda CyclicBarrier.

Bantuan sinkronisasi yang memungkinkan serangkaian utas menunggu satu sama lain untuk mencapai titik penghalang bersama. CyclicBarriers berguna dalam program yang melibatkan sekelompok thread berukuran tetap yang terkadang harus menunggu satu sama lain. Penghalang ini disebut siklik karena dapat digunakan kembali setelah utas menunggu dilepaskan.

Kemungkinan lain adalah CountDownLatchobjek, yang berguna untuk situasi sederhana: karena Anda mengetahui jumlah utas sebelumnya, Anda menginisialisasinya dengan hitungan yang relevan, dan meneruskan referensi objek ke setiap utas.
Setelah menyelesaikan tugasnya, setiap utas memanggil CountDownLatch.countDown()yang mengurangi penghitung internal. Utas utama, setelah memulai yang lain, harus melakukan CountDownLatch.await()panggilan pemblokiran. Ini akan dirilis segera setelah penghitung internal mencapai 0.

Perhatikan bahwa dengan benda ini, InterruptedExceptionbisa juga dilempar.

Tunggu / blokir Thread Main sampai beberapa thread lain menyelesaikan pekerjaannya.

Seperti yang @Ravindra babudikatakan itu dapat dicapai dengan berbagai cara, tetapi menunjukkan dengan contoh.

• java.lang.Thread. join () ^{Sejak: 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 ^{Sejak: 1.5}

  • .countDown() «Mengurangi jumlah kelompok kait.
  • .await() «Metode menunggu memblokir hingga hitungan saat ini mencapai nol.

  Jika Anda membuatnya latchGroupCount = 4maka countDown()harus dipanggil 4 kali untuk menghitung 0. Jadi, itu await()akan melepaskan utas pemblokiran.

  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.
  }
  

Contoh kode kelas Berulir LatchTask. Untuk menguji penggunaan pendekatan joiningThreads(); dan latchThreads();dari metode utama.

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 Bantuan sinkronisasi yang memungkinkan satu set utas menunggu satu sama lain untuk mencapai titik penghalang yang sama. CyclicBarriers berguna dalam program yang melibatkan sekelompok utas berukuran tetap yang terkadang harus menunggu satu sama lain. Penghalang ini disebut siklik karena dapat digunakan kembali setelah utas menunggu dilepaskan.

  CyclicBarrier barrier = new CyclicBarrier(3);
  barrier.await();
  

  Misalnya, lihat kelas Concurrent_ParallelNotifyies ini .

• Kerangka kerja Executer: kita dapat menggunakan ExecutorService untuk membuat kumpulan utas, dan melacak kemajuan tugas asinkron dengan Future.

  • submit(Runnable), submit(Callable)yang mengembalikan Objek Masa Depan. Dengan menggunakan future.get()fungsi kita dapat memblokir thread utama hingga thread yang bekerja menyelesaikan tugasnya.

  • invokeAll(...) - mengembalikan daftar objek Future yang dengannya Anda bisa mendapatkan hasil dari eksekusi setiap Callable.

Temukan contoh penggunaan Interfaces Runnable, Callable with Executor framework.

@Lihat juga

• Cari tahu utas masih hidup?

Anda lakukan

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

Setelah perulangan for ini, Anda dapat memastikan semua utas telah menyelesaikan tugasnya.

Simpan objek Thread ke dalam beberapa koleksi (seperti List atau Set), lalu lakukan loop melalui koleksi setelah thread dimulai dan panggil join () pada Thread.

Anda dapat menggunakan metode gabungan Threadf # untuk tujuan ini.

Meskipun tidak relevan dengan masalah OP, jika Anda tertarik pada sinkronisasi (lebih tepatnya, rendez-vous) dengan satu utas, Anda dapat menggunakan Exchanger

Dalam kasus saya, saya perlu menghentikan sementara utas induk sampai utas anak melakukan sesuatu, misalnya menyelesaikan inisialisasi. Sebuah CountDownLatch juga bekerja dengan baik.

Layanan pelaksana dapat digunakan untuk mengelola banyak utas termasuk status dan penyelesaian. Lihat http://programmingexamples.wikidot.com/executorservice

coba ini, akan berhasil.

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

        }

   }

Saya memiliki masalah yang sama dan akhirnya menggunakan Java 8 parallelStream.

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

Ini sangat sederhana dan mudah dibaca. Di balik layar itu menggunakan kumpulan gabungan garpu JVM default yang berarti akan menunggu semua utas selesai sebelum melanjutkan. Untuk kasus saya, ini adalah solusi yang rapi, karena itu adalah satu-satunya parallelStream dalam aplikasi saya. Jika Anda memiliki lebih dari satu parallelStream yang berjalan secara bersamaan, harap baca tautan di bawah ini.

Informasi selengkapnya tentang aliran paralel di sini .

Saya membuat metode pembantu kecil untuk menunggu beberapa Thread selesai:

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

Jawaban yang ada mengatakan bisa join()tiap utas.

Tetapi ada beberapa cara untuk mendapatkan larik / daftar utas:

• Tambahkan Thread ke dalam daftar saat pembuatan.
• Gunakan ThreadGroupuntuk mengelola utas.

Kode berikut akan menggunakan ThreadGruoppendekatan tersebut. Ini membuat grup terlebih dahulu, kemudian ketika membuat setiap utas menentukan grup dalam konstruktor, nanti bisa mendapatkan array utas melaluiThreadGroup.enumerate()

Kode

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

Utas utama akan menunggu semua utas dalam grup selesai.

Gunakan ini di utas utama Anda: while (! Executor.isTerminated ()); Letakkan baris kode ini setelah memulai semua utas dari layanan pelaksana. Ini hanya akan memulai utas utama setelah semua utas yang dimulai oleh pelaksana selesai. Pastikan untuk memanggil executor.shutdown (); sebelum loop di atas.