2 CountDownLatch¶

CountDownLatch¶

La API de concurrencia de Java nos proporciona la clase CountDownLatch, que permite a uno o más hilos esperar hasta que un conjunto de operaciones son ejecutadas.

El constructor de esta clase recibe un parámetro entero que indica el número de operaciones que los hilos deberán esperar antes de poder continuar su ejecución. Se trata, por tanto, de un sistema de control de avance (pestillo) que sólo se abre cuando su cuenta atrás es 0.

Cuando un hilo ejecuta el método await() de un objeto de esta clase, el hilo se suspende hasta que sean completadas el número de operaciones indicadas al crear el objeto CountDownLatch.

El método await(timeout, timeUnit) está sobrecargado de manera que le podemos pasar el tiempo máximo que el hilo estará esperando a que finalice la cuenta atrás del objeto CountDownLatch, transcurrido el cual será reactivado y continuará su ejecución aunque la cuenta atrás no haya llegado a 0, retornando el valor false. Si el contador interno llega al valor 0 el método retornará el valor true.

Como la mayoría de los métodos bloqueantes, await() y await(timeout, timeUnit) lanzarán la excepción InterruptedException si el hilo es interrumpido mientras estaba esperando en dichos métodos para poder avanzar, o si ya había sido marcado para interrupción antes de ejecutarlos, reactivando inmediatamente el hilo correspondiente.

Cada vez que termina la ejecución de una de las operaciones gestionadas que afecten a la cuenta atrás del CountDownLatch, debe llamarse al método countDown() del mismo, que decrementa el contador interno del pestillo en una unidad. Cuando este contador llega a 0, la clase reactiva todos los hilos que estuvieran suspendidos por el método await() o await(timeout, timeUnit), que continúan su ejecución.

Debemos tener en cuenta que un hilo puede gestionar más de una operación que afecte a la cuenta atrás del CountDownLatch y por tanto llamar a su método countDown() más de una vez. Podríamos por tanto decir que un CountDownLatch no sincroniza hilos en un determinado punto de ejecución, sino que bloquea el avance de hilos hasta que ocurra un determinado número de eventos, independientemente de en qué hilos se generen dichos eventos.

No hay ninguna manera de reinicializar el contador interno del objeto CountDownLatch o de modificar directamente su valor. Una vez que el contador se inicializa al crear el objeto, el único método que modifica su valor es countDown(), como vimos en el párrafo anterior. Si que tenemos un método informativo, getCount() que retorna el valor actual del contador interno del CountDownLatch.

El objeto CountDownLatch admite un único uso. Una vez que el contador interno llega a 0 todas las llamadas a sus métodos no tienen efecto alguno, de manera que los hilos que los llaman continúan su ejecución normalmente. Se tendría que crear un nuevo objeto de dicha clase para llevar a cabo la misma sincronización de nuevo. Se dice, por tanto, que los CountDownLatch no son cíclicos, sino que son de un solo "tiro" (one-shot).

Otro aspecto curioso es que la clase CountDownLatch no implementa ningún modo de justicia, y que no existe una versión de await() en la que no sea posible interrumpir al hilo.

Proyecto CountDownLatch¶

En este proyecto simularemos una reunión de delegaciones de países para la que cada país aporta un número de participantes. La reunión no puede comenzar hasta que no haya un quorum de al menos 10 participantes, independiente de qué país procedan. Para controlar que la reunión no comienza hasta que no haya quorum usaremos una CountDownLatch. Una vez haya comenzado la reunión las delegaciones podrán realizar propuestas para la misma.

import java.util.concurrent.ThreadLocalRandom;

public class Main {

    private static final int QUORUM = 10;

    public static void main(String[] args) {
        Meeting meeting = new Meeting(QUORUM);
        new Thread(meeting).start();
        for (int i = 0; i < 10; i++) {
            new Thread(new Country(
                    ThreadLocalRandom.current().nextInt(5) + 1,
                    meeting, "Country #" + i)).start();
        }
    }

}

import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.CountDownLatch;

public class Meeting implements Runnable {

    private final CountDownLatch countDownLatch;
    private final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("HH:mm:ss");

    public Meeting(int quorum) {
        countDownLatch = new CountDownLatch(quorum);
    }

    @Override
    public void run() {
        System.out.printf("%s -> Waiting for quorum to start the meeting\n",
                LocalTime.now().format(dateTimeFormatter));
        try {
            countDownLatch.await();
            System.out.printf("%s -> We have quorum. Meeting started...\n",
                    LocalTime.now().format(dateTimeFormatter));
        } catch (InterruptedException e) {
            System.out.println("Meeting has been interrupted while waiting to have quorum");
        }
    }

    public void join(int participants, String countryName) {
        System.out.printf("%s -> %d participants from %s have joined the meeting\n",
                LocalTime.now().format(dateTimeFormatter), participants, countryName);
        for (int i = 0; i < participants; i++) {
            countDownLatch.countDown();
        }
    }

    public void propose(String countryName) {
        System.out.printf("%s -> Delegation from %s has made some proposals\n",
                LocalTime.now().format(dateTimeFormatter), countryName);
    }

}

import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

public class Country implements Runnable {

    private final int participants;
    private final String countryName;
    private final Meeting meeting;

    public Country(int participants, Meeting meeting, String countryName) {
        if (participants < 1 || meeting == null || countryName == null) {
            throw new IllegalArgumentException();
        }
        this.participants = participants;
        this.countryName = countryName;
        this.meeting = meeting;
    }

    @Override
    public void run() {
        try {
            goToMeeting();
        } catch (InterruptedException e) {
            System.out.println("I've been interrupted while going to the meeting");
            return;
        }
        meeting.join(participants, countryName);
        try {
            makeProposals();
        } catch (InterruptedException e) {
            System.out.println("I've been interrupted while making proposals");
        }
    }

    private void goToMeeting() throws InterruptedException {
        TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(5));
    }

    private void makeProposals() throws InterruptedException {
        TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(3));
        meeting.propose(countryName);
    }

}

Si ejecutamos el programa veremos que hasta que no haya quorum no comenzará la reunión.