espere hasta que todos los hilos terminen su trabajo en java

Estoy escribiendo una aplicación que tiene 5 subprocesos que obtienen información de la web simultáneamente y llenan 5 campos diferentes en una clase de búfer.
Necesito validar los datos del búfer y almacenarlos en una base de datos cuando todos los subprocesos terminaron su trabajo.
¿Cómo puedo hacer esto (recibir una alerta cuando todos los hilos terminen su trabajo)?

El enfoque que tomo es usar un ExecutorService para administrar grupos de subprocesos.

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.

Puedes joina los hilos. La unión se bloquea hasta que se completa el hilo.

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

Tenga en cuenta que joinarroja un InterruptedException. Tendrá que decidir qué hacer si eso sucede (por ejemplo, intente cancelar los otros hilos para evitar que se realice un trabajo innecesario).

Eche un vistazo a varias soluciones.

1. join()La API se ha introducido en las primeras versiones de Java. Algunas buenas alternativas están disponibles con este paquete concurrente desde el lanzamiento de JDK 1.5.

2. ExecutorService # invokeAll ()

   Ejecuta las tareas asignadas, devolviendo una lista de Futuros con su estado y resultados cuando todo está terminado.

   Consulte esta pregunta de SE relacionada para ver un ejemplo de código:

   ¿Cómo usar invokeAll () para permitir que todos los grupos de subprocesos hagan su tarea?

3. CountDownLatch

   Una ayuda de sincronización que permite que uno o más subprocesos esperen hasta que se complete un conjunto de operaciones que se están realizando en otros subprocesos.

   Un CountDownLatch se inicializa con un recuento determinado. Los métodos de espera se bloquean hasta que el recuento actual llega a cero debido a las invocaciones del countDown()método, después de lo cual se liberan todos los subprocesos en espera y cualquier invocación posterior del retorno de espera inmediatamente. Este es un fenómeno de una sola vez: el recuento no se puede restablecer. Si necesita una versión que restablezca el conteo, considere usar CyclicBarrier .

   Consulte esta pregunta para conocer el uso de CountDownLatch

   ¿Cómo esperar a que un hilo genere su propio hilo?

4. ForkJoinPool o newWorkStealingPool () en Ejecutores

5. Iterar a través de todos los objetos futuros creados después de enviar aExecutorService

Además de lo Thread.join()sugerido por otros, java 5 introdujo el marco ejecutor. Allí no trabajas con Threadobjetos. En su lugar, enviar sus Callableo Runnablelos objetos a un ejecutor. Hay un ejecutor especial que está destinado a ejecutar múltiples tareas y devolver sus resultados fuera de orden. Ese es el ExecutorCompletionService:

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

Luego, puede llamar repetidamente take()hasta que no haya más Future<?>objetos para devolver, lo que significa que todos están completados.

Otra cosa que puede ser relevante, dependiendo de su escenario es CyclicBarrier.

Una ayuda de sincronización que permite que un conjunto de hilos esperen unos a otros para alcanzar un punto de barrera común. CyclicBarriers son útiles en programas que involucran un grupo de subprocesos de tamaño fijo que ocasionalmente deben esperar entre sí. La barrera se llama cíclica porque se puede reutilizar después de que se liberan los hilos en espera.

Otra posibilidad es el CountDownLatchobjeto, que es útil para situaciones simples: como conoces de antemano el número de subprocesos, lo inicializas con el recuento relevante y pasas la referencia del objeto a cada subproceso.
Al completar su tarea, cada hilo llama, lo CountDownLatch.countDown()que disminuye el contador interno. El hilo principal, después de iniciar todos los demás, debe hacer elCountDownLatch.await() llamada de bloqueo. Se liberará tan pronto como el contador interno llegue a 0.

Tenga en cuenta que con este objeto también InterruptedExceptionse puede lanzar un.

Tú lo haces

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

Después de este ciclo for, puede estar seguro de que todos los subprocesos han terminado su trabajo.

Espere / bloquee el hilo principal hasta que otros hilos completen su trabajo.

Como se @Ravindra babudijo se puede lograr de varias formas, pero mostrando con ejemplos.

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

  • .countDown() «Disminuye el recuento del grupo de pestillos.
  • .await() «Los métodos de espera se bloquean hasta que la cuenta actual llega a cero.

  Si lo creó latchGroupCount = 4, countDown()debería llamarse 4 veces para que la cuenta sea 0. Entonces, eso await()liberará los hilos de bloqueo.

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

Código de ejemplo de la clase Threaded LatchTask. Para probar el uso del enfoque joiningThreads(); y latchThreads();del método principal.

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 Una ayuda de sincronización que permite que un conjunto de subprocesos esperen unos a otros para alcanzar un punto de barrera común. CyclicBarriers son útiles en programas que involucran una parte de subprocesos de tamaño fijo que ocasionalmente deben esperar entre sí. La barrera se llama cíclica porque se puede reutilizar después de que se liberan los hilos en espera.

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

  Por ejemplo, consulte esta clase Concurrent_ParallelNotifyies .

• Marco de ejecución: podemos usar ExecutorService para crear un grupo de subprocesos y rastrear el progreso de las tareas asincrónicas con Future.

  • submit(Runnable), submit(Callable)que devuelven Future Object. Al usar la future.get()función, podemos bloquear el hilo principal hasta que los hilos de trabajo completen su trabajo.

  • invokeAll(...) - devuelve una lista de objetos Future a través de los cuales puede obtener los resultados de las ejecuciones de cada invocable.

Encuentre un ejemplo de uso de interfaces ejecutables, invocables con el marco de ejecución.

@Ver también

• ¿Descubrir que el hilo sigue vivo?

Almacene los objetos Thread en alguna colección (como una Lista o un Conjunto), luego recorra la colección una vez que se inicien los hilos y llame a join () en los hilos.

Puede utilizar el método de unión Threadf # para este propósito.

Aunque no es relevante para el problema de OP, si está interesado en la sincronización (más precisamente, un encuentro) con exactamente un hilo, puede usar un Intercambiador

En mi caso, necesitaba pausar el hilo principal hasta que el hilo secundario hiciera algo, por ejemplo, completara su inicialización. Un CountDownLatch también funciona bien.

Se puede usar un servicio ejecutor para administrar múltiples subprocesos, incluido el estado y la finalización. Consulte http://programmingexamples.wikidot.com/executorservice

prueba esto, funcionará.

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

        }

   }

Tuve un problema similar y terminé usando Java 8 paraleloStream.

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

Es super simple y legible. Detrás de escena, está utilizando el grupo de unión de bifurcación de JVM predeterminado, lo que significa que esperará a que finalicen todos los hilos antes de continuar. Para mi caso, fue una buena solución, porque era el único paralelo en mi aplicación. Si tiene más de una transmisión en paralelo ejecutándose simultáneamente, lea el enlace a continuación.

Más información sobre corrientes paralelas aquí .

Las respuestas existentes dijeron que podría join()cada hilo.

Pero hay varias formas de obtener la matriz / lista de subprocesos:

• Agregue el hilo en una lista en la creación.
• Úselo ThreadGrouppara administrar los hilos.

El siguiente código utilizará el ThreadGruopenfoque. Primero crea un grupo, luego, cuando crea cada hilo, especifique el grupo en el constructor, más tarde podría obtener la matriz de hilos a través deThreadGroup.enumerate()

Código

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

El hilo principal esperará a que finalicen todos los hilos del grupo.

Creé un pequeño método auxiliar para esperar a que terminen algunos subprocesos:

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

Use esto en su hilo principal: while (! Ejecutor.isTerminated ()); Coloque esta línea de código después de iniciar todos los hilos del servicio ejecutor. Esto solo iniciará el hilo principal después de que todos los hilos iniciados por los ejecutores hayan finalizado. Asegúrese de llamar a executeor.shutdown (); antes del bucle anterior.