¿Cómo se usa CountDownLatch en Java Multithreading?

¿Alguien puede ayudarme a comprender qué CountDownLatches Java y cuándo usarlo?

No tengo una idea muy clara de cómo funciona este programa. Según tengo entendido, los tres subprocesos comienzan a la vez y cada subproceso llamará a CountDownLatch después de 3000 ms. Entonces la cuenta regresiva disminuirá uno por uno. Después de que el pestillo se vuelve cero, el programa imprime "Completado". Tal vez la forma en que entendí es incorrecta.

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

class Processor implements Runnable {
    private CountDownLatch latch;

    public Processor(CountDownLatch latch) {
        this.latch = latch;
    }

    public void run() {
        System.out.println("Started.");

        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        latch.countDown();
    }
}

// ------------------------------------------------ -----

public class App {

    public static void main(String[] args) {

        CountDownLatch latch = new CountDownLatch(3); // coundown from 3 to 0

        ExecutorService executor = Executors.newFixedThreadPool(3); // 3 Threads in pool

        for(int i=0; i < 3; i++) {
            executor.submit(new Processor(latch)); // ref to latch. each time call new Processes latch will count down by 1
        }

        try {
            latch.await();  // wait until latch counted down to 0
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("Completed.");
    }

}

Sí, entendiste correctamente. CountDownLatchfunciona en el principio de cierre, el hilo principal esperará hasta que la puerta esté abierta. Un subproceso espera n subprocesos, especificados al crear el CountDownLatch.

Cualquier hilo, generalmente el hilo principal de la aplicación, cuyas llamadas CountDownLatch.await()esperarán hasta que el recuento llegue a cero o sea interrumpido por otro hilo. Todos los otros subprocesos deben realizar una cuenta regresiva llamando CountDownLatch.countDown()una vez que estén completos o listos.

Tan pronto como el recuento llegue a cero, el hilo de espera continúa. Una de las desventajas / ventajas de CountDownLatches que no es reutilizable: una vez que el recuento llega a cero, ya no puede usarlo CountDownLatch.

Editar:

Úselo CountDownLatchcuando un subproceso (como el subproceso principal) requiere esperar a que uno o más subprocesos se completen, antes de que pueda continuar el procesamiento.

Un ejemplo clásico de uso CountDownLatchen Java es una aplicación Java central del lado del servidor que usa arquitectura de servicios, donde múltiples servicios son provistos por múltiples hilos y la aplicación no puede comenzar a procesarse hasta que todos los servicios se hayan iniciado con éxito.

La pregunta de PS OP tiene un ejemplo bastante sencillo, así que no incluí uno.

CountDownLatchen Java es un tipo de sincronizador que permite Thread esperar uno o más Threadsegundos antes de comenzar a procesar.

CountDownLatchfunciona según el principio de cierre, el hilo esperará hasta que la puerta esté abierta. Un subproceso espera el nnúmero de subprocesos especificados durante la creación CountDownLatch.

p.ej final CountDownLatch latch = new CountDownLatch(3);

Aquí ponemos el contador a 3.

Cualquier hilo, generalmente el hilo principal de la aplicación, cuyas llamadas CountDownLatch.await()esperarán hasta que el recuento llegue a cero o sea interrumpido por otro Thread. Todos los otros subprocesos deben realizar una cuenta regresiva llamando CountDownLatch.countDown()una vez que se hayan completado o estén listos para el trabajo. Tan pronto como el recuento llegue a cero, la Threadespera comenzará a correr.

Aquí el recuento se reduce por CountDownLatch.countDown()método.

El Threadque llama al await()método esperará hasta que el recuento inicial llegue a cero.

Para hacer el recuento cero, otros hilos necesitan llamar al countDown()método. Una vez que el recuento llegue a cero, el hilo que invocó el await()método se reanudará (comenzará su ejecución).

La desventaja CountDownLatches que no es reutilizable: una vez que el recuento se vuelve cero, ya no se puede usar.

NikolaB lo explicó muy bien, sin embargo, un ejemplo sería útil de entender, así que aquí hay un ejemplo simple ...

 import java.util.concurrent.*;


  public class CountDownLatchExample {

  public static class ProcessThread implements Runnable {

    CountDownLatch latch;
    long workDuration;
    String name;

    public ProcessThread(String name, CountDownLatch latch, long duration){
        this.name= name;
        this.latch = latch;
        this.workDuration = duration;
    }


    public void run() {
        try {
            System.out.println(name +" Processing Something for "+ workDuration/1000 + " Seconds");
            Thread.sleep(workDuration);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(name+ "completed its works");
        //when task finished.. count down the latch count...

        // basically this is same as calling lock object notify(), and object here is latch
        latch.countDown();
    }
}


public static void main(String[] args) {
    // Parent thread creating a latch object
    CountDownLatch latch = new CountDownLatch(3);

    new Thread(new ProcessThread("Worker1",latch, 2000)).start(); // time in millis.. 2 secs
    new Thread(new ProcessThread("Worker2",latch, 6000)).start();//6 secs
    new Thread(new ProcessThread("Worker3",latch, 4000)).start();//4 secs


    System.out.println("waiting for Children processes to complete....");
    try {
        //current thread will get notified if all chidren's are done 
        // and thread will resume from wait() mode.
        latch.await();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

    System.out.println("All Process Completed....");

    System.out.println("Parent Thread Resuming work....");



     }
  }

Se usa cuando queremos esperar a más de un hilo para completar su tarea. Es similar unirse en hilos.

Donde podemos usar CountDownLatch

Considere un escenario en el que tenemos requisitos donde tenemos tres hilos "A", "B" y "C" y queremos comenzar el hilo "C" solo cuando los hilos "A" y "B" completan o completan parcialmente su tarea.

Se puede aplicar al escenario de TI del mundo real

Considere un escenario en el que el gerente dividió los módulos entre los equipos de desarrollo (A y B) y quiere asignarlo al equipo de control de calidad para probar solo cuando ambos equipos completen su tarea.

public class Manager {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(2);
        MyDevTeam teamDevA = new MyDevTeam(countDownLatch, "devA");
        MyDevTeam teamDevB = new MyDevTeam(countDownLatch, "devB");
        teamDevA.start();
        teamDevB.start();
        countDownLatch.await();
        MyQATeam qa = new MyQATeam();
        qa.start();
    }   
}

class MyDevTeam extends Thread {   
    CountDownLatch countDownLatch;
    public MyDevTeam (CountDownLatch countDownLatch, String name) {
        super(name);
        this.countDownLatch = countDownLatch;       
    }   
    @Override
    public void run() {
        System.out.println("Task assigned to development team " + Thread.currentThread().getName());
        try {
                Thread.sleep(2000);
        } catch (InterruptedException ex) {
                ex.printStackTrace();
        }
    System.out.println("Task finished by development team Thread.currentThread().getName());
            this.countDownLatch.countDown();
    }
}

class MyQATeam extends Thread {   
    @Override
    public void run() {
        System.out.println("Task assigned to QA team");
        try {
                Thread.sleep(2000);
        } catch (InterruptedException ex) {
            ex.printStackTrace();
        }
        System.out.println("Task finished by QA team");
    }
}

La salida del código anterior será:

Tarea asignada al equipo de desarrollo devB

Tarea asignada al equipo de desarrollo devA

Tarea finalizada por el equipo de desarrollo devB

Tarea finalizada por el equipo de desarrollo devA

Tarea asignada al equipo de control de calidad

Tarea finalizada por el equipo de control de calidad

Aquí aguardan () espera del procedimiento para la bandera countdownlatch para convertirse en 0, y cuenta atrás () método decrementa bandera countdownlatch por 1.

Limitación de JOIN: el ejemplo anterior también se puede lograr con JOIN, pero JOIN no se puede usar en dos escenarios:

1. Cuando usamos ExecutorService en lugar de la clase Thread para crear hilos.
2. Modifique el ejemplo anterior donde el Administrador desea transferir el código al equipo de control de calidad tan pronto como el Desarrollo complete su tarea del 80%. Significa que CountDownLatch nos permite modificar la implementación que se puede utilizar para esperar otro subproceso para su ejecución parcial.

CoundDownLatch le permite hacer que un hilo espere hasta que todos los otros hilos hayan terminado su ejecución.

El pseudocódigo puede ser:

// Main thread starts
// Create CountDownLatch for N threads
// Create and start N threads
// Main thread waits on latch
// N threads completes there tasks are returns
// Main thread resume execution

Un buen ejemplo de cuándo usar algo como esto es con Java Simple Serial Connector, accediendo a los puertos seriales. Por lo general, escribirá algo en el puerto y, de forma asíncrona, en otro hilo, el dispositivo responderá en un SerialPortEventListener. Por lo general, querrás pausar después de escribir en el puerto para esperar la respuesta. Manejar los bloqueos de hilo para este escenario manualmente es extremadamente complicado, pero usar Countdownlatch es fácil. Antes de pensar que puedes hacerlo de otra manera, ¡ten cuidado con las condiciones de carrera en las que nunca pensaste!

Pseudocódigo:

CountDownLatch latch;
void writeData() { 
   latch = new CountDownLatch(1);
   serialPort.writeBytes(sb.toString().getBytes())
   try {
      latch.await(4, TimeUnit.SECONDS);
    } catch (InterruptedException e) {
   }
}
class SerialPortReader implements SerialPortEventListener {
    public void serialEvent(SerialPortEvent event) {
        if(event.isRXCHAR()){//If data is available
            byte buffer[] = serialPort.readBytes(event.getEventValue());
            latch.countDown();
         }
     }
}

Si agrega alguna depuración después de su llamada a latch.countDown (), esto puede ayudarlo a comprender mejor su comportamiento.

latch.countDown();
System.out.println("DONE "+this.latch); // Add this debug

La salida mostrará que el conteo se está decrementando. Este 'conteo' es efectivamente el número de tareas Ejecutables (objetos del procesador) que ha comenzado contra las cuales countDown () no ha sido invocado y, por lo tanto, está bloqueado el hilo principal en su llamada a latch.await ().

DONE java.util.concurrent.CountDownLatch@70e69696[Count = 2]
DONE java.util.concurrent.CountDownLatch@70e69696[Count = 1]
DONE java.util.concurrent.CountDownLatch@70e69696[Count = 0]

De la documentación de Oracle sobre CountDownLatch :

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

A CountDownLatchse inicializa con un recuento dado. Los awaitmétodos 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 de espera de retorno inmediatamente. Este es un fenómeno de una sola vez: el recuento no se puede restablecer.

CountDownLatch es una herramienta de sincronización versátil y se puede usar para varios propósitos.

Un CountDownLatchinicializado con un recuento de uno sirve como un simple pestillo de encendido / apagado, o puerta: todos los hilos que invocan esperan en la puerta hasta que se abre por un hilo que invoca countDown ().

Un CountDownLatchinicializado a N puede usarse para hacer que un subproceso espere hasta que N subprocesos hayan completado alguna acción, o alguna acción se haya completado N veces.

public void await()
           throws InterruptedException

Hace que el hilo actual espere hasta que el pestillo haya regresado a cero, a menos que el hilo se interrumpa.

Si el recuento actual es cero, este método vuelve inmediatamente.

public void countDown()

Disminuye la cuenta del pestillo, liberando todos los hilos de espera si la cuenta llega a cero.

Si el recuento actual es mayor que cero, entonces se reduce. Si el nuevo recuento es cero, todos los hilos en espera se vuelven a habilitar para fines de programación de hilos.

Explicación de tu ejemplo.

1. Ha establecido el recuento como 3 para latchvariable

   CountDownLatch latch = new CountDownLatch(3);

2. Has pasado este latchhilo compartido a Trabajador:Processor

3. Se han enviado tres Runnableinstancias deProcessorExecutorService executor

4. El subproceso principal ( App) está esperando que el recuento se vuelva cero con la siguiente declaración

    latch.await();  

5. Processor el hilo duerme durante 3 segundos y luego disminuye el valor de conteo con latch.countDown()

6. La primera Processinstancia cambiará el recuento de pestillos como 2 después de su finalización debido alatch.countDown() .

7. La segunda Processinstancia cambiará el recuento de bloqueos como 1 después de que se complete debido a latch.countDown().

8. La tercera Processinstancia cambiará el recuento de bloqueo como 0 después de que se complete debido a latch.countDown().

9. El conteo cero en el pestillo hace que el hilo principal Appsalga deawait

10. El programa de aplicación imprime esta salida ahora: Completed

Este ejemplo de Java Doc me ayudó a comprender los conceptos claramente:

class Driver { // ...
  void main() throws InterruptedException {
    CountDownLatch startSignal = new CountDownLatch(1);
    CountDownLatch doneSignal = new CountDownLatch(N);

    for (int i = 0; i < N; ++i) // create and start threads
      new Thread(new Worker(startSignal, doneSignal)).start();

    doSomethingElse();            // don't let run yet
    startSignal.countDown();      // let all threads proceed
    doSomethingElse();
    doneSignal.await();           // wait for all to finish
  }
}

class Worker implements Runnable {
  private final CountDownLatch startSignal;
  private final CountDownLatch doneSignal;
  Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
     this.startSignal = startSignal;
     this.doneSignal = doneSignal;
  }
  public void run() {
     try {
       startSignal.await();
       doWork();
       doneSignal.countDown();
     } catch (InterruptedException ex) {} // return;
  }

  void doWork() { ... }
}

Interpretación visual:

Evidentemente, CountDownLatchpermite que un subproceso (aquí Driver) espere hasta que se terminen varios subprocesos en ejecución (aquí Worker) con su ejecución.

Como se menciona en JavaDoc ( https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/CountDownLatch.html ), CountDownLatch es una ayuda de sincronización, introducida en Java 5. Aquí la sincronización no significa restringir el acceso a una sección crítica. Sino más bien secuencias de acciones de diferentes hilos. El tipo de sincronización logrado a través de CountDownLatch es similar al de Join. Suponga que hay un subproceso "M" que necesita esperar a que otros subprocesos de trabajo "T1", "T2", "T3" completen sus tareas Antes de Java 1.5, la forma en que esto se puede hacer es que M ejecute el siguiente código

    T1.join();
    T2.join();
    T3.join();

El código anterior se asegura de que el hilo M reanude su trabajo después de que T1, T2, T3 completen su trabajo. T1, T2, T3 pueden completar su trabajo en cualquier orden. Lo mismo se puede lograr a través de CountDownLatch, donde T1, T2, T3 y el hilo M comparten el mismo objeto CountDownLatch.
Solicitudes "M": countDownLatch.await();
donde como "T1", "T2", "T3" countDownLatch.countdown();

Una desventaja con el método de unión es que M tiene que saber acerca de T1, T2, T3. Si hay un nuevo subproceso de trabajo T4 agregado más tarde, M también debe tenerlo en cuenta. Esto se puede evitar con CountDownLatch. Después de la implementación, la secuencia de acción sería [T1, T2, T3] (el orden de T1, T2, T3 podría ser de todos modos) -> [M]

El mejor ejemplo en tiempo real para countDownLatch explicado en este enlace CountDownLatchExample

package practice;

import java.util.concurrent.CountDownLatch;

public class CountDownLatchExample {

    public static void main(String[] args) throws InterruptedException {
        CountDownLatch c= new CountDownLatch(3);  // need to decrements the count (3) to zero by calling countDown() method so that main thread will wake up after calling await() method 
        Task t = new Task(c);
        Task t1 = new Task(c);
        Task t2 = new Task(c);
        t.start();
        t1.start();
        t2.start();
        c.await(); // when count becomes zero main thread will wake up 
        System.out.println("This will print after count down latch count become zero");
    }
}

class Task extends Thread{
    CountDownLatch c;

    public Task(CountDownLatch c) {
        this.c = c;
    }

    @Override
    public void run() {
        try {
            System.out.println(Thread.currentThread().getName());
            Thread.sleep(1000);
            c.countDown();   // each thread decrement the count by one 
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}