Comment le CountDownLatch est-il utilisé dans le multithreading Java ?

Question

Quelqu'un peut-il m'aider à comprendre ce que Java CountDownLatch est et quand l'utiliser ?

Je n'ai pas une idée très claire du fonctionnement de ce programme. Si je comprends bien, les trois threads démarrent en même temps et chaque thread appelle CountDownLatch après 3000ms. Le compte à rebours sera donc décrémenté un par un. Une fois que le verrou est à zéro, le programme imprime "Completed". Peut-être la façon dont j'ai compris est incorrecte.

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.");
    }

}

Answer 1

Comme mentionné dans la JavaDoc ( https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/CountDownLatch.html ), CountDownLatch est une aide à la synchronisation, introduite dans Java 5. Ici, la synchronisation ne signifie pas la restriction de l'accès à une section critique. Mais plutôt de séquencer les actions des différents threads. Le type de synchronisation obtenu par CountDownLatch est similaire à celui de Join. Supposons qu'il existe un thread "M" qui doit attendre que d'autres threads "T1", "T2", "T3" aient terminé leurs tâches. Avant Java 1.5, la façon dont cela peut être fait est que M exécute le code suivant

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

Le code ci-dessus fait en sorte que le thread M reprenne son travail après que T1, T2, T3 ait terminé son travail. T1, T2, T3 peuvent terminer leur travail dans n'importe quel ordre. La même chose peut être réalisée par CountDownLatch, où T1, T2, T3 et le thread M partagent le même objet CountDownLatch.
Demandes "M" : countDownLatch.await();
alors que "T1", "T2", "T3" font countDownLatch.countdown();

Un inconvénient de la méthode de jonction est que M doit connaître T1, T2, T3. Si un nouveau fil de travail T4 est ajouté plus tard, M doit également en être informé. Ceci peut être évité avec CountDownLatch. Après l'implémentation, la séquence d'action serait [T1,T2,T3] (l'ordre de T1,T2,T3 peut être différent) -> [M].

Answer 2

Un bon exemple d'utilisation de ce type de dispositif est le connecteur Java Simple Serial Connector, qui permet d'accéder aux ports série. En général, vous écrivez quelque chose sur le port et, de manière asynchrone, sur un autre thread, le périphérique répond à un SerialPortEventListener. Typiquement, vous voudrez faire une pause après avoir écrit sur le port pour attendre la réponse. Gérer manuellement les verrous des threads pour ce scénario est extrêmement délicat, mais utiliser Countdownlatch est facile. Avant d'aller penser que vous pouvez le faire d'une autre manière, faites attention aux conditions de course auxquelles vous n'avez jamais pensé !

Pseudocode :

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

Answer 3

Si vous ajoutez un débogage après votre appel à latch.countDown(), cela peut vous aider à mieux comprendre son comportement.

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

La sortie montrera que le compte est décrémenté. Ce "compte" est en fait le nombre de tâches exécutables (objets Processeur) que vous avez lancées et par rapport auxquelles countDown() a été utilisé. no a été invoqué et est donc bloqué par le thread principal lors de son appel à 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]

Answer 4

Dans la documentation d'oracle sur CountDownLatch :

Une aide à la synchronisation qui permet à un ou plusieurs threads d'attendre la fin d'un ensemble d'opérations effectuées dans d'autres threads.

A CountDownLatch est initialisé avec un nombre donné. Le site await bloquent jusqu'à ce que le compte courant atteigne zéro à cause des invocations de la méthode countDown() après quoi tous les threads en attente sont libérés et toutes les invocations ultérieures de await reviennent immédiatement. Il s'agit d'un phénomène unique - le compte ne peut pas être remis à zéro.

Un CountDownLatch est un outil de synchronisation polyvalent et peut être utilisé à de nombreuses fins.

A CountDownLatch initialisé avec un compte de un sert de simple loquet on/off, ou porte : tous les threads invoquant await attendent à la porte jusqu'à ce qu'elle soit ouverte par un thread invoquant countDown().

A CountDownLatch initialisé à N peut être utilisé pour faire attendre un thread jusqu'à ce que N threads aient terminé une action, ou qu'une action ait été terminée N fois.

public void await()
           throws InterruptedException

Fait attendre le fil d'exécution actuel jusqu'à ce que le verrou ait décompté jusqu'à zéro, à moins que le fil d'exécution ne soit interrompu.

Si le compte actuel est égal à zéro, cette méthode revient immédiatement.

public void countDown()

Diminue le compte du latch, libérant tous les threads en attente si le compte atteint zéro.

Si le compte courant est supérieur à zéro, il est décrémenté. Si le nouveau compte est égal à zéro, tous les threads en attente sont réactivés pour la planification des threads.

Explication de votre exemple.

1. Vous avez fixé le compte à 3 pour latch variable

   CountDownLatch latch = new CountDownLatch(3);

2. Vous avez réussi ce partage latch au fil de travail : Processor

3. Trois Runnable instances de Processor ont été soumis à ExecutorService executor

4. Fil conducteur ( App ) attend que le compte devienne nul avec l'instruction suivante

    latch.await();  

5. Processor Le thread dort pendant 3 secondes et ensuite il décrémente la valeur du compte avec latch.countDown()

6. Premier Process l'instance changera le nombre de latchs en 2 après son achèvement en raison de latch.countDown() .

7. Deuxièmement Process l'instance changera le nombre de latchs en 1 après son achèvement en raison de latch.countDown() .

8. Troisièmement Process l'instance changera le nombre de latchs en 0 après son achèvement en raison de latch.countDown() .

9. Le comptage du zéro sur le latch provoque le thread principal App pour sortir de await

10. Le programme d'application imprime cette sortie maintenant : Completed

Answer 5

Cet exemple tiré de Doc Java m'a aidé à comprendre clairement les concepts :

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() { ... }
}

Interprétation visuelle :

De toute évidence, CountDownLatch permet d'un fil (ici Driver ) pour attendre qu'un groupe de fils en cours d'exécution (ici Worker ) ont terminé leur exécution.