Tutoriel sur la programmation concurrente en Java

La spécification de Java prévoit à sa base des mécanismes de moniteur. C'est ainsi que l'on retrouve le mot clé synchronized qui permet de surveiller (et protéger) l'utilisation d'une instance :

//com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.api_native.Synchronized
Object verrou = new Object();

new Thread() {
  public void run() {
    Logger.println("Pause");
    Thread.sleep(200);
    Logger.println("Verrouillage");
    synchronized (verrou) {
      Logger.println("Verrou posé");
    }
    Logger.println("Verrou libéré");
  }
}.start();

new Thread() {
  public void run() {
    Logger.println("Verrouillage");
    synchronized (verrou) {
      Logger.println("Verrou posé");
      Logger.println("Pause");
      Thread.sleep(1500);
    }
    Logger.println("Verrou libéré");
  }
}.start();

00:00:00.020 [Thread-0       ] Pause
00:00:00.020 [Thread-1       ] Verrouillage
00:00:00.027 [Thread-1       ] Verrou posé
00:00:00.028 [Thread-1       ] Pause
00:00:00.227 [Thread-0       ] Verrouillage
00:00:01.528 [Thread-1       ] Verrou libéré
00:00:01.528 [Thread-0       ] Verrou posé
00:00:01.529 [Thread-0       ] Verrou libéré

Lorsqu'un thread entre dans un bloc synchronized, il acquiert un « moniteur » sur l'objet concerné. Il est possible de vérifier si un thread détient le moniteur via la méthode Thread.holdsLock(Object) : 

//com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.api_native.ThreadHoldsLock
Object verrou = new Object();

Logger.println("Avant  : %b", Thread.holdsLock(verrou));
synchronized (verrou) {
  Logger.println("Dedans : %b", Thread.holdsLock(verrou));
}
Logger.println("Après  : %b", Thread.holdsLock(verrou));

00:00:00.018 [main           ] Avant  : false
00:00:00.026 [main           ] Dedans : true
00:00:00.026 [main           ] Après  : false

Les moniteurs appliquent le principe de « réentrance ». Ainsi un thread qui a déjà acquis un moniteur ne sera pas bloqué s'il demande à nouveau l'acquisition du même moniteur.

De pair avec les blocs synchronized, il existe les méthodes wait et notify pour respectivement mettre en attente et réveiller un thread. Celles-ci ne peuvent être appelées que depuis un thread qui détient le moniteur, dans le cas contraire une IllegalMonitorStateException sera levée :

//com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.api_native.IllegalMonitorState
Object verrou = new Object();
Runnable tache = new Runnable() {
  public void run() {
    boolean tryWait;
    try {
      verrou.wait(100);
      tryWait = true;
    } catch (IllegalMonitorStateException e) {
      tryWait = false;
    }
    boolean tryNotify;
    try {
      verrou.notify();
      tryNotify = true;
    } catch (IllegalMonitorStateException e) {
      tryNotify = false;
    }
    Logger.println("Synchronized (%-5b): wait: %-5b, notifty: %-5b", Thread.holdsLock(verrou), tryWait, tryNotify);
  }
};

tache.run();
synchronized (verrou) {
  tache.run();
}

00:00:00.019 [main           ] Synchronized (false): wait: false, notifty: false
00:00:00.126 [main           ] Synchronized (true ): wait: true , notifty: true

Les méthodes wait permettent de mettre le thread courant en attente (limitée ou indéterminée) et libèrent le moniteur de l'objet concerné (mais pas les autres moniteurs !). Ainsi un autre thread est autorisé à détenir le verrou (et entrer dans un bloc synchronized). Un thread demeure en attente jusqu'à : 

• l'expiration du délai d'attente (si spécifié et strictement positif) ;

  Wait - temps d'attente (code)

  Sélectionnez

  //com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.api_native.WaitTempsAttente
  new Thread() {
    public void run() {
      Object verrou = new Object();
      Logger.println("Synchronisation");
      synchronized (verrou) {
        Logger.println("Attente - début");
        verrou.wait(1_000);
        Logger.println("Attente - fin");
      }
    }
  }.start();

  Wait - temps d'attente (console)

  Sélectionnez

  00:00:00.019 [Thread-0       ] Synchronisation
  00:00:00.026 [Thread-0       ] Attente - début
  00:00:01.027 [Thread-0       ] Attente - fin

• son réveil suite à un appel à la méthode notify de l'objet sur lequel il est en attente ;

  Wait - notify (code)

  Sélectionnez

  //com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.api_native.WaitNotify
  Object verrou = new Object();
  abstract class Tache implements Runnable {
    public void run() {
      Logger.println("Synchronisation");
      synchronized (verrou) {
        action();
      }
    }
  
    abstract void action();
  };
  Tache patient = new Tache() {
    void action() {
      Logger.println("Attente - début");
      verrou.wait(5_000);
      Logger.println("Attente - fin");
    }
  };
  Tache reveil = new Tache() {
    void action() {
      Logger.println("Pause");
      Thread.sleep(500);
      Logger.println("Réveil");
      verrou.notify();
    }
  };
  new Thread(patient, "Patient").start();
  new Thread(reveil , "Réveil").start();

  Wait - notify (console)

  Sélectionnez

  00:00:00.020 [Patient        ] Synchronisation
  00:00:00.020 [Réveil         ] Synchronisation
  00:00:00.027 [Patient        ] Attente - début
  00:00:00.028 [Réveil         ] Pause
  00:00:00.528 [Réveil         ] Réveil
  00:00:00.528 [Patient        ] Attente - fin

• son interruption.

  Wait - interruption (code)

  Sélectionnez

  //com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.api_native.WaitInterruption
  Object verrou = new Object();
  Thread patient = new Thread("Patient") {
    public void run() {
      Logger.println("Synchronisation");
      synchronized (verrou) {
        try {
          Logger.println("Attente - début");
          verrou.wait(5_000);
          Logger.println("Attente - fin");
        } catch (InterruptedException e) {
          Logger.printStackTrace(e);
        }
      }
    }
  };
  Thread reveil = new Thread("Reveil") {
    public void run() {
      Logger.println("Pause");
      Thread.sleep(500);
      Logger.println("Reveil");
      patient.interrupt();
    }
  };
  patient.start();
  reveil.start();

  Wait - interruption (console)

  Sélectionnez

  00:00:00.022 [Patient        ] Synchronisation
  00:00:00.022 [Reveil         ] Pause
  00:00:00.030 [Patient        ] Attente - début
  00:00:00.531 [Reveil         ] Reveil
  00:00:00.561 [Patient        ] java.lang.InterruptedException
  java.lang.InterruptedException
    at java.lang.Object.wait(Native Method)
    at com.developpez.lmauzaize.java.concurrence.api_native.WaitInterruption$1.run(WaitInterruption.java:15)

Les blocs synchronized permettent de protéger très simplement l'accès à une référence tandis que les méthodes wait et notify permettent de mettre en attente un thread jusqu'à ce qu'une condition soit remplie. Voici un exemple simple permettant l'échange de données entre deux threads :

//com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.api_native.ThreadEchange
class Tâche extends Thread {
  Object[] échange;
  int src;
  public Tâche(Object[] échange, int src) {
    super("Tâche-" + src);
    this.échange = échange;
    this.src = src;
  }

  public void run() {
    long pause = 500 * src;
    Logger.println("Pause - %04dms", pause);
    sleep(pause);

    synchronized (échange) {
      // Envoi
      int dst = (src + 1) % échange.length;
      échange[dst] = getName();
      échange.notifyAll();
      Logger.println("Envoi vers Tâche-%d", dst);

      // Réception
      while (échange[src] == null) {
        échange.wait(5_000);
      }
      Logger.println("Réception de %-7s", échange[src]);
    }
  }
};

Object[] échange = new Object[4];
for (int i = 0; i < échange.length; i++) {
  new Tâche(échange, i).start();
}

00:00:00.020 [Tâche-0        ] Pause - 0000ms
00:00:00.020 [Tâche-3        ] Pause - 1500ms
00:00:00.020 [Tâche-1        ] Pause - 0500ms
00:00:00.020 [Tâche-2        ] Pause - 1000ms
00:00:00.027 [Tâche-0        ] Envoi vers Tâche-1
00:00:00.528 [Tâche-1        ] Envoi vers Tâche-2
00:00:00.528 [Tâche-1        ] Réception de Tâche-0
00:00:01.028 [Tâche-2        ] Envoi vers Tâche-3
00:00:01.028 [Tâche-2        ] Réception de Tâche-1
00:00:01.527 [Tâche-3        ] Envoi vers Tâche-0
00:00:01.527 [Tâche-3        ] Réception de Tâche-2
00:00:01.528 [Tâche-0        ] Réception de Tâche-3

volatile est un mot clé qui pose souvent beaucoup de problèmes sur sa compréhension. Contrairement à synchronized, ce mot clé s'applique à une variable. Il consiste à indiquer à la VM que cette variable est susceptible d'être modifiée et lue par plusieurs threads et donc que la VM doit faire son possible pour maintenir la cohérence des lectures de la variable. Cela implique donc que chaque lecture et chaque écriture se fassent en mémoire centrale, et non pas uniquement sur la pile du thread et les caches. En effet, la VM est libre de mettre en cache ou sur la pile du thread des variables non locales. Du fait de cette liberté (en fonction des implémentations et des versions), il est difficile de démontrer l'effet de ce mot clé.

D'autres problèmes résolus par ce mot clé sont :

• Word tearing : certains processeurs n'étant pas capables d'écrire un seul octet en mémoire, le processeur réécrit alors un ensemble d'octets. C'est notamment le cas si vous manipulez des tableaux de byte. Il est parfaitement acceptable d'un point de vue de la spécification que plusieurs threads écrivent à des indices différents d'un même tableau sans risque de concurrence. Cependant de tels processeurs causent des erreurs sans le mot clé volatile. Voici un morceau de code permettant de tester votre processeur :

  Word Tearing

  Sélectionnez

  //com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.api_native.WordTearing
  byte[] bytes = new byte[10_000];
  class Tache extends Thread {
    private int index;
    public Tache(int index) {
      super("Tache-" + index);
      this.index = index;
    }
    @Override
    public void run() {
      bytes[index] = (byte) index;
    }
  };
  Tache[] taches = new Tache[bytes.length];
  for (int i = 0; i < taches.length; i++) {
    taches[i] = new Tache(i);
  }
  for (Tache tache : taches) {
    tache.start();
  }
  for (Tache tache : taches) {
    tache.join();
  }
  int erreur = 0;
  for (int i = 0; i < bytes.length; i++) {
    byte val = (byte) i;
    if (bytes[i] != val) {
      erreur++;
      Logger.println("bytes[%02d]=%02d", i, bytes[i]);
    }
  }
  Logger.println("Résultat: %02d erreur(s)", erreur);

• Écriture non atomique des types 64-bits : sur des architectures 32-bits, les implémentations de la VM sont libres d'écrire les deux « mots » des types 64-bits (long, double) en une seule ou deux fois. Dans ce cas, il est possible que l'une des deux parties ait été modifiée entre l'écriture de la première partie et celle de la deuxième.

Les Semaphores permettent de limiter le nombre de threads accédant à une ressource. Un Semaphore s'initialise avec un certain nombre de permissions. L'acquisition (acquire) attend la disponibilité du nombre de permissions demandées et décrémente le nombre disponible d'autant. La libération (release) rend disponible le nombre de permissions correspondant.

//com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.semaphore.SemaphoreDemo
class Action {
  Semaphore semaphore = new Semaphore(10);

  public void disponibilité() {
    Logger.println("%-15s %02d", "disponibilité", semaphore.availablePermits());
  }
  public void acquisition(int count) {
    Logger.println("%-15s %02d", "acquisition", count);
    semaphore.acquire(count);
    disponibilité();
  }
  public void libération(int count) {
    Logger.println("%-15s %02d", "libération", count);
    semaphore.release(count);
    disponibilité();
  }
};
Action action = new Action();

action.disponibilité();
action.libération(5);
action.acquisition(15);

new Thread("Acquisiteur") {
  public void run() {
    action.acquisition(15);
  }
}.start();
new Thread("Libérateur") {
  public void run() {
    sleep(500);
    action.disponibilité();
    for (int i = 0; i < 3; i++) {
      action.libération(5);
      sleep(500);
    }
  }
}.start();

00:00:00.022 [main           ] disponibilité   10
00:00:00.031 [main           ] libération      05
00:00:00.031 [main           ] disponibilité   15
00:00:00.031 [main           ] acquisition     15
00:00:00.032 [main           ] disponibilité   00
00:00:00.033 [Acquisiteur    ] acquisition     15
00:00:00.533 [Libérateur     ] disponibilité   00
00:00:00.533 [Libérateur     ] libération      05
00:00:00.534 [Libérateur     ] disponibilité   05
00:00:01.035 [Libérateur     ] libération      05
00:00:01.035 [Libérateur     ] disponibilité   10
00:00:01.536 [Libérateur     ] libération      05
00:00:01.537 [Libérateur     ] disponibilité   15
00:00:01.537 [Acquisiteur    ] disponibilité   00

//com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.semaphore.SemaphoreDeadlock
Semaphore semaphore = new Semaphore(1);
Thread thread = new Thread("Deadlock") {
  public void run() {
    Logger.println("début");
    try {
      Logger.println("acquisition #1");
      Logger.println("disponibilité: %02d", semaphore.availablePermits());
      semaphore.acquire();
      Logger.println("acquisition #2");
      Logger.println("disponibilité: %02d", semaphore.availablePermits());
      semaphore.acquire();
      Logger.println("acquisitions terminées");
      Logger.println("disponibilité: %02d", semaphore.availablePermits());
    } catch (InterruptedException e) {
      Logger.printStackTrace(e);
      Logger.println("disponibilité %02d", semaphore.availablePermits());
    }
    Logger.println("fin");
  }
};
thread.start();
thread.join(2_000);
if (thread.isAlive()) {
  Logger.println("disponibilité: %02d", semaphore.availablePermits());
  Logger.println("Thread toujours vivant, deadlock présumé");
  thread.interrupt();
  semaphore.release();
}

00:00:00.040 [Deadlock       ] début
00:00:00.049 [Deadlock       ] acquisition #1
00:00:00.049 [Deadlock       ] disponibilité: 01
00:00:00.049 [Deadlock       ] acquisition #2
00:00:00.049 [Deadlock       ] disponibilité: 00
00:00:01.998 [main           ] disponibilité: 00
00:00:01.998 [main           ] Thread toujours vivant, deadlock présumé
00:00:02.001 [Deadlock       ] java.lang.InterruptedException
java.lang.InterruptedException
  at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:998)
  at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
  at java.util.concurrent.Semaphore.acquire(Semaphore.java:312)
  at com.developpez.lmauzaize.java.concurrence.semaphore.SemaphoreDeadlock$1.run(SemaphoreDeadlock.java:20)

00:00:02.001 [Deadlock       ] disponibilité 01
00:00:02.002 [Deadlock       ] fin

Par défaut, un Semaphore est impartial (ou non équitable, unfair), c'est-à-dire qu'il ne respecte pas l'ordre de mise en attente des threads. À la libération des permissions, celles disponibles seront affectées de manière arbitraire aux threads en attente. Si vous voulez respecter l'ordre FIFO (First In First Out), vous devez utiliser le constructeur à deux paramètres avec le paramètre fair à true.

//com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.semaphore.SemaphoreEquité
boolean équité    = false;
int     size      = 2_250;
int     iteration = 200;

Semaphore semaphore = new Semaphore(1, équité);
List<Thread> actuals   = new ArrayList<>(size);
List<Thread> expecteds = new ArrayList<>(size);
Runnable tache = new Runnable() {
  public void run() {
    semaphore.acquire();
    actuals.add(Thread.currentThread());
    semaphore.release();
  }
};

int erreur = 0;
int j = 0;
for (; j < iteration; j++) {

  // Créé de nouveaux threads et vérifie qu'il soit mis en attente avant de passer à la suite
  semaphore.acquire();
  for (int i = 0; i < size; i++) {
    Thread thread = new Thread(tache);
    thread.start();
    expecteds.add(thread);
    while (expecteds.size() != semaphore.getQueueLength()) {
      Thread.yield();
    }
  }
  
  // Débloque les threads et attend qu'ils soient tous terminés.
  semaphore.release();
  for (Thread thread : expecteds) {
    thread.join();
  }

  // Vérifie les différences d'ordre
  int difference = 0;
  for (int i = 0; i < size; i++) {
    Thread expected = expecteds.get(i);
    Thread actual   = actuals.get(i);
    if (expected != actual) {
      difference++;
      Logger.println("Expected: %-15s Actual: %-15s", expected.getName(), actual.getName());
    }
  }
  actuals.clear();
  expecteds.clear();
  Logger.println("Itération %03d : %03d erreur(s)", j, difference);
  erreur += difference;
}
Logger.println("%03d erreur(s) en %03d itérations", erreur, iteration);

Il existe également deux autres méthodes intéressantes qui sont tryAcquire et acquireUninterruptibly. La première permet d'essayer d'acquérir la ressource (avec éventuellement un timeout), ainsi on peut continuer le traitement (et même réagir) plutôt que d'attendre la disponibilité de la ressource. La deuxième méthode est tout à l'opposé en réalisant une attente non interruptible ; contrairement à une boucle sur acquire, l'interruption du thread n'affecte pas sa position dans la file d'attente (si l'équité est activée).

Un CountDownLatch (« Loquet à compte à rebours ») permet de bloquer l'exécution d'un ou plusieurs thread jusqu'à ce que le compteur atteigne zéro.

//com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.countDownLatch.CountDownLatchDemo
int nbThread = 5;
CountDownLatch loquet = new CountDownLatch(nbThread);
Runnable tache = new Runnable() {
  public void run() {
    Logger.println("Début");
    loquet.countDown();
    loquet.await();
    Logger.println("Fin");
  }
};
for (int i = 0; i < nbThread; i++) {
  new Thread(tache).start();
  Thread.sleep(500);
}

00:00:00.020 [Thread-0       ] Début
00:00:00.499 [Thread-1       ] Début
00:00:00.999 [Thread-2       ] Début
00:00:01.499 [Thread-3       ] Début
00:00:01.999 [Thread-4       ] Début
00:00:02.000 [Thread-4       ] Fin
00:00:02.000 [Thread-3       ] Fin
00:00:02.000 [Thread-1       ] Fin
00:00:02.000 [Thread-2       ] Fin
00:00:02.000 [Thread-0       ] Fin

Il est en quelque sorte l'opposé du Semaphore. Au lieu de protéger une ressource, il permet de synchroniser l'avancement d'un (ou plusieurs threads) à différentes étapes. On peut par exemple :

• attendre que tous les threads soient initialisés (voir exemple ci-dessus),

• attendre la fin de toutes les sous-tâches,

  CountDownLatch - attente des sous-tâches (code)

  Sélectionnez

  //com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.countDownLatch.CountDownLatchAttenteSousTaches
  long[] pauses = { 4_000, 5_000, 2_000, 1_000 };
  CountDownLatch latch = new CountDownLatch(pauses.length);
  class Tache extends Thread {
    long pause;
    Tache(long pause) {
      this.pause = pause;
    }
    public void run() {
      Logger.println("Pause %04dms", pause);
      Thread.sleep(pause);
      Logger.println("Fin");
      latch.countDown();
    }
  };
  
  for (long pause : pauses) {
    new Tache(pause).start();
  }
  latch.await();
  Logger.println("Fin du processus");

  CountDownLatch - attente des sous-tâches (console)

  Sélectionnez

  00:00:00.020 [Thread-3       ] Pause 1000ms
  00:00:00.020 [Thread-1       ] Pause 5000ms
  00:00:00.020 [Thread-0       ] Pause 4000ms
  00:00:00.020 [Thread-2       ] Pause 2000ms
  00:00:01.027 [Thread-3       ] Fin
  00:00:02.028 [Thread-2       ] Fin
  00:00:04.028 [Thread-0       ] Fin
  00:00:05.028 [Thread-1       ] Fin
  00:00:05.028 [main           ] Fin du processus

• synchroniser le déroulement des étapes.

  CountDownLatch - étapes (code)

  Sélectionnez

  //com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.countDownLatch.CountDownLatchEtape
  // Signaux
  CountDownLatch pif  = new CountDownLatch(1);
  CountDownLatch paf  = new CountDownLatch(1);
  CountDownLatch pouf = new CountDownLatch(1);
  
  
  class Tache extends Thread {
    public Tache(String nom) {
      super(nom);
    }
  
    // Signaux précédents
    CountDownLatch[] avant;
    Tache avant(CountDownLatch... avant) {
      this.avant = avant;
      return this;
    }
    // Signal à envoyer
    CountDownLatch signal;
    Tache signal(CountDownLatch signal) {
      this.signal = signal;
      return this;
    }
    // Signaux suivants
    CountDownLatch[] après;
    Tache après(CountDownLatch... après) {
      this.après = après;
      return this;
    }
  
    public void run() {
      // avant
      for (CountDownLatch loquet : avant) {
        loquet.await();
      }
      // signal
      Thread.sleep(1_000);
      Logger.println(getName());
      signal.countDown();
      // après
      for (CountDownLatch loquet : après) {
        loquet.await();
      }
      Logger.println("Fin");
    }
  }
  new Tache("Pif  !").avant(        ).signal(pif ).après(paf, pouf).start();
  new Tache("Paf  !").avant(pif     ).signal(paf ).après(pouf     ).start();
  new Tache("Pouf !").avant(pif, paf).signal(pouf).après(         ).start();

  CountDownLatch - étapes (console)

  Sélectionnez

  00:00:00.040 [Pif  !         ] Pif  !
  00:00:01.053 [Paf  !         ] Paf  !
  00:00:02.053 [Pouf !         ] Pouf !
  00:00:02.053 [Pouf !         ] Fin
  00:00:02.053 [Pif  !         ] Fin
  00:00:02.053 [Paf  !         ] Fin

Une CyclicBarrier permet (comme un CountDownLatch) de bloquer un (ou plusieurs) thread(s) jusqu'à ce que toutes les tâches aient atteint la barrière. En effet, contrairement à un CountDownLatch, le compteur ne se décrémente qu'avec la méthode de synchronisation (await). Pour imager, vous pouvez voir la barrière comme un sas, la porte d'entée est toujours ouverte et les tâches peuvent s'entasser jusqu'à ce que le sas soit rempli. Dès lors que le sas est rempli, la porte d'entrée se ferme et la porte de sortie s'ouvre.

//com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.cyclicBarrier.CyclicBarrierDemo
int nbThread = 5;
CyclicBarrier barrière = new CyclicBarrier(nbThread + 1);
Runnable tache = new Runnable() {
  public void run() {
    Logger.println("Début");
    int index = nbThread - barrière.await();
    Logger.println("Fin (%d)", index);
  }
};
for (int i = 0; i < nbThread; i++) {
  new Thread(tache).start();
  Thread.sleep(500);
}
barrière.await();

00:00:00.019 [Thread-0       ] Début
00:00:00.499 [Thread-1       ] Début
00:00:00.999 [Thread-2       ] Début
00:00:01.499 [Thread-3       ] Début
00:00:01.999 [Thread-4       ] Début
00:00:02.499 [Thread-0       ] Fin (0)
00:00:02.499 [Thread-2       ] Fin (2)
00:00:02.499 [Thread-4       ] Fin (4)
00:00:02.499 [Thread-3       ] Fin (3)
00:00:02.499 [Thread-1       ] Fin (1)

Une des fonctionnalités intéressantes des CyclicBarrier est son modèle « tout-ou-rien ». Si une tentative de synchronisation échoue alors la barrière est « cassée », et toutes les attentes sont annulées. Une barrière est cassée si :

• un des threads en attente est interrompu. Dans ce cas, le thread interrompu lève une InterruptedException et tous les autres en attente lèvent une BrokenBarrierException,

  CyclicBarrier - interruption (code)

  Sélectionnez

  //com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.cyclicBarrier.CyclicBarrierInterruption
  CyclicBarrier barrière = new CyclicBarrier(10);
  Runnable tache = new Runnable() {
    public void run() {
      try {
        barrière.await();
      } catch (Exception e) {
        Logger.println("1. %s", e);
      }
      Logger.println("Est cassée=%s", barrière.isBroken());
      try {
        barrière.await();
      } catch (Exception e) {
        Logger.println("2. %s", e);
      }
    }
  };
  Thread thread = new Thread(tache, "Interruption");
  new Thread(tache).start();
  new Thread(tache).start();
  new Thread(tache).start();
  thread.start();
  Thread.sleep(2_000);
  thread.interrupt();
  Thread.sleep(1_000);
  Logger.println("Est cassée=%s", barrière.isBroken());

  CyclicBarrier - interruption (console)

  Sélectionnez

  00:00:00.040 [Thread-0       ] 1. java.util.concurrent.BrokenBarrierException
  00:00:00.040 [Interruption   ] 1. java.lang.InterruptedException
  00:00:00.040 [Thread-2       ] 1. java.util.concurrent.BrokenBarrierException
  00:00:00.040 [Thread-1       ] 1. java.util.concurrent.BrokenBarrierException
  00:00:00.050 [Thread-1       ] Est cassée=true
  00:00:00.050 [Thread-2       ] Est cassée=true
  00:00:00.050 [Interruption   ] Est cassée=true
  00:00:00.049 [Thread-0       ] Est cassée=true
  00:00:00.051 [Thread-0       ] 2. java.util.concurrent.BrokenBarrierException
  00:00:00.051 [Interruption   ] 2. java.util.concurrent.BrokenBarrierException
  00:00:00.051 [Thread-2       ] 2. java.util.concurrent.BrokenBarrierException
  00:00:00.050 [Thread-1       ] 2. java.util.concurrent.BrokenBarrierException
  00:00:00.998 [main           ] Est cassée=true

• un des threads en attente voit son délai expiré. Dans ce cas, le thread ayant expiré lève une TimeoutException et tous les autres en attente lèvent une BrokenBarrierException,

  CyclicBarrier - expiration (code)

  Sélectionnez

  //com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.cyclicBarrier.CyclicBarrierExpiration
  CyclicBarrier barrière = new CyclicBarrier(10);
  class Tache extends Thread {
    private long attente;
    Tache(String nom, long attente) {
      super(nom);
      this.attente = attente;
    }
  
    public void run() {
      try {
        barrière.await(attente, TimeUnit.MILLISECONDS);
      } catch (Exception e) {
        Logger.println("1. %s", e);
      }
      Logger.println("Est cassée=%s", barrière.isBroken());
      try {
        barrière.await();
      } catch (Exception e) {
        Logger.println("2. %s", e);
      }
    }
  };
  new Tache("Patient #1", 60_000).start();
  new Tache("Patient #2", 60_000).start();
  new Tache("Patient #3", 60_000).start();
  new Tache("Expiration",  2_000).start();
  Thread.sleep(3_000);
  Logger.println("Est cassée=%s", barrière.isBroken());

  CyclicBarrier - expiration (console)

  Sélectionnez

  00:00:00.035 [Patient #2     ] 1. java.util.concurrent.BrokenBarrierException
  00:00:00.035 [Expiration     ] 1. java.util.concurrent.TimeoutException
  00:00:00.035 [Patient #3     ] 1. java.util.concurrent.BrokenBarrierException
  00:00:00.035 [Patient #1     ] 1. java.util.concurrent.BrokenBarrierException
  00:00:00.046 [Patient #3     ] Est cassée=true
  00:00:00.046 [Expiration     ] Est cassée=true
  00:00:00.046 [Patient #2     ] Est cassée=true
  00:00:00.047 [Expiration     ] 2. java.util.concurrent.BrokenBarrierException
  00:00:00.047 [Patient #3     ] 2. java.util.concurrent.BrokenBarrierException
  00:00:00.046 [Patient #1     ] Est cassée=true
  00:00:00.047 [Patient #2     ] 2. java.util.concurrent.BrokenBarrierException
  00:00:00.048 [Patient #1     ] 2. java.util.concurrent.BrokenBarrierException
  00:00:00.998 [main           ] Est cassée=true

• un thread réinitialise la barrière. Dans ce cas, tous les threads en attente lèvent une BrokenBarrierException,

  CyclicBarrier - réinitialisation (code)

  Sélectionnez

  //com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.cyclicBarrier.CyclicBarrierRéinitialisation
  int nbThread = 4;
  CyclicBarrier barrière = new CyclicBarrier(nbThread+1);
  Runnable tache = new Runnable() {
    public void run() {
      try {
        barrière.await();
      } catch (Exception e) {
        Logger.println("1. %s", e);
      }
      Logger.println("Est cassée=%s", barrière.isBroken());
      try {
        barrière.await();
      } catch (Exception e) {
        Logger.println("2. %s", e);
      }
    }
  };
  for (int i = 0; i < nbThread; i++) new Thread(tache).start();
  Thread.sleep(2_000);
  barrière.reset();
  Logger.println("Est cassée=%s", barrière.isBroken());
  barrière.await();

  CyclicBarrier - réinitialisation (console)

  Sélectionnez

  00:00:00.036 [Thread-3       ] 1. java.util.concurrent.BrokenBarrierException
  00:00:00.036 [Thread-1       ] 1. java.util.concurrent.BrokenBarrierException
  00:00:00.036 [Thread-0       ] 1. java.util.concurrent.BrokenBarrierException
  00:00:00.036 [Thread-2       ] 1. java.util.concurrent.BrokenBarrierException
  00:00:00.036 [main           ] Est cassée=false
  00:00:00.045 [Thread-2       ] Est cassée=false
  00:00:00.045 [Thread-0       ] Est cassée=false
  00:00:00.045 [Thread-1       ] Est cassée=false
  00:00:00.044 [Thread-3       ] Est cassée=false

Si la barrière est « consommée » normalement (c'est-à-dire sans être « cassée »), elle est prête à être réutilisée immédiatement. Enfin un dernier élément concernant les CyclicBarrier consiste à exécuter une action (une et une seule fois) dès que la barrière se lève. L'action est alors exécutée par le dernier thread entrant dans la barrière.

//com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.cyclicBarrier.CyclicBarrierAction
int nbThread = 4;
CyclicBarrier barrière = new CyclicBarrier(nbThread, () -> Logger.println("Action"));
Runnable tache = new Runnable() {
  public void run() {
    Logger.println("Attente");
    barrière.await();
    Logger.println("Fin");
  }
};
for (int i = 0; i < nbThread; i++) {
  new Thread(tache).start();
  Thread.sleep(500);
}

00:00:00.018 [Thread-0       ] Attente
00:00:00.499 [Thread-1       ] Attente
00:00:00.999 [Thread-2       ] Attente
00:00:01.499 [Thread-3       ] Attente
00:00:01.500 [Thread-3       ] Action
00:00:01.500 [Thread-3       ] Fin
00:00:01.500 [Thread-2       ] Fin
00:00:01.500 [Thread-1       ] Fin
00:00:01.500 [Thread-0       ] Fin

Les Phasers (ou « moniteur pas-à-pas ») permettent de délimiter des phases de synchronisation. Par exemple, vous pouvez souhaiter réaliser différentes tâches, mais que certaines se synchronisent à différentes étapes. Un phaser se caractérise par deux éléments principaux : un numéro de phase et des parties inscrites. Pour s'inscrire, il faut appeler la méthode register. Lorsqu'une partie a terminé son travail, elle le signale via la méthode arrive. Lorsque toutes les parties ont atteint la phase en cours, le numéro de phase est alors incrémenté.

//com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.phaser.PhaserDemo
Phaser phaser = new Phaser() {
  {
    println("<init>");
  }

  private void println(String action) {
    Logger.println("{%-10s} phase=%02d  registred=%02d arrived=%02d unarrived=%02d", action, getPhase(), getRegisteredParties(), getArrivedParties(), getUnarrivedParties());
  }

  public int register() {
    int register = super.register();
    println("register");
    return register;
  }

  public int arrive() {
    int arrive = super.arrive();
    println("arrive");
    return arrive;
  }

  protected boolean onAdvance(int phase, int registeredParties) {
    Logger.println("{%-10s} phase=%02d  registred=%02d", "onAdvance", phase, registeredParties);
    return super.onAdvance(phase, registeredParties);
  }
};
phaser.register();
phaser.register();
phaser.arrive();
phaser.arrive();

00:00:00.021 [main           ] {<init>    } phase=00  registred=00 arrived=00 unarrived=00
00:00:00.030 [main           ] {register  } phase=00  registred=01 arrived=00 unarrived=01
00:00:00.031 [main           ] {register  } phase=00  registred=02 arrived=00 unarrived=02
00:00:00.033 [main           ] {arrive    } phase=00  registred=02 arrived=01 unarrived=01
00:00:00.034 [main           ] {onAdvance } phase=00  registred=02
00:00:00.034 [main           ] {arrive    } phase=01  registred=02 arrived=00 unarrived=02

Sans attente, ce type de synchronisation n'est pas plus utile qu'un simple compteur. Il existe trois types d'attente :

• arriveAndAwaitAdvance : indique qu'une partie a atteint la fin de la phase et attend que les autres parties aient également atteint la fin de la phase ;
• awaitAdvance : attend que la phase spécifiée en paramètre soit terminée. Si la phase passée en paramètre ne correspond pas à la phase courante, la méthode se termine immédiatement ;
• awaitAdvanceInterruptibly : même chose que précédemment, excepté que les interruptions ne sont pas capturées.

Voici un exemple en revisitant un autre donné précédemment :

//com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.phaser.PhaserEtape
String[] parties = { "Pif !", "Paf !", "Pouf !" };
Phaser phaser = new Phaser();
class Tache extends Thread {
  private int étape;
  public Tache(int étape) {
    super(parties[étape]);
    this.étape = étape;
  }

  public void run() {
    for (int i = 0; i < étape; i++) {
      phaser.arriveAndAwaitAdvance();
    }
    Thread.sleep(1_000);
    Logger.println("[%d] %s", phaser.arriveAndAwaitAdvance(), getName());
    for (int i = étape; i < parties.length; i++) {
      phaser.arriveAndAwaitAdvance();
    }
  }
}
phaser.bulkRegister(parties.length);
for (int i = parties.length-1; i >= 0; i--) {
  new Tache(i).start();
  Thread.sleep(200);
}

00:00:00.020 [Pif !          ] [1] Pif !
00:00:00.999 [Paf !          ] [2] Paf !
00:00:01.999 [Pouf !         ] [3] Pouf !

Une autre caractéristique d'un phaser est son statut terminé. Lorsqu'un phaser est terminé, toutes les méthodes retournent immédiatement (et le cas échéant, un numéro de phase négatif). Pour terminer un phaser, il suffit d'appeler la méthode forceTermination. Celle-ci provoque également la fin de la mise en attente des toutes les parties. Une fois terminé, un phaser ne peut pas être réinitialisé pour être réutilisé, il faudra créer une nouvelle instance. La terminaison d'un phaser permet de lancer un processus de « recouvrement » si l'une des parties a rencontré un problème :

//com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.phaser.PhaserTerminaison
Phaser phaser = new Phaser();
OfInt séquence = IntStream.iterate(0, i -> i+1).iterator();
class Standard extends Thread {
  int numéro = séquence.nextInt();
  int sleep = numéro * 200;
  {
    setName(getClass().getSimpleName() + "-" + numéro);
    phaser.register();
  }

  void pause() throws InterruptedException {
    Logger.println("Pause % 4dms", sleep);
    sleep(sleep);
  }

  void arrive() throws InterruptedException {
    pause();
    int étape = phaser.arrive();
    Logger.println("Étape %02d terminée", étape);
    étape = phaser.awaitAdvance(étape);
    Logger.println("Étape %02d atteinte", étape);
  }

  public void run() {
    arrive();
    arrive();
    Logger.println("Fin");
  }
}
class Erreur extends Standard {
  public void run() {
    arrive();
    pause();
    Logger.println("Erreur rencontrée");
    phaser.forceTermination();
    Logger.println("Fin");
  }
};
for (Standard tâche : Arrays.asList(new Standard(), new Erreur(), new Standard())) {
  tâche.start();
}

00:00:00.020 [Erreur-1       ] Pause  200ms
00:00:00.020 [Standard-2     ] Pause  400ms
00:00:00.020 [Standard-0     ] Pause    0ms
00:00:00.028 [Standard-0     ] Étape 00 terminée
00:00:00.228 [Erreur-1       ] Étape 00 terminée
00:00:00.428 [Standard-2     ] Étape 00 terminée
00:00:00.428 [Erreur-1       ] Étape 01 atteinte
00:00:00.428 [Standard-0     ] Étape 01 atteinte
00:00:00.428 [Erreur-1       ] Pause  200ms
00:00:00.428 [Standard-2     ] Étape 01 atteinte
00:00:00.429 [Standard-0     ] Pause    0ms
00:00:00.429 [Standard-2     ] Pause  400ms
00:00:00.430 [Standard-0     ] Étape 01 terminée
00:00:00.629 [Erreur-1       ] Erreur rencontrée
00:00:00.629 [Erreur-1       ] Fin
00:00:00.629 [Standard-0     ] Étape -2147483647 atteinte
00:00:00.629 [Standard-0     ] Fin
00:00:00.830 [Standard-2     ] Étape -2147483647 terminée
00:00:00.830 [Standard-2     ] Étape -2147483647 atteinte
00:00:00.830 [Standard-2     ] Fin

Un Phaser possède des limites à connaître. La première concerne le nombre et le numéro des phases. Il n'y a pas de limites au nombre de phases, en revanche, il faut être averti qu'une fois la phase Integer.MAX_VALUE atteinte, la suivante aura le numéro 0.

La seconde limite concerne le nombre de parties. Celui-ci est limité à 65 535. Cependant, une fonctionnalité intéressante des Phasers est de pouvoir former une hiérarchie avec une infinité de niveaux. Pour créer un nouveau niveau, il suffit d'utiliser le constructeur Phaser(Phaser parent). Tous les Phasers d'une hiérarchie possèdent le même numéro d'étape. Une hiérarchie permet d'avoir une infinité de parties et de réduire les contentions. Chaque enfant agit comme une partie dès que son propre nombre de parties est différent de zéro et se désinscrit dès lors qu'il n'a plus de partie.

//com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.phaser.PhaserNiveaux
// Gère un "pool" nommé de phaser et trace les actions
// L'état de tous les phasers est tracé après chaque action
class PhaserLogger {
  void créer(String nom);
  void créer(String parent, String nom);
  void register(String nom);
  void arriveAndDeregister(String nom);
}

PhaserLogger logger = new PhaserLogger();
logger.créer("racine");
logger.register("racine");
logger.créer("racine", "enfant");
logger.register("enfant");
logger.register("enfant");
logger.arriveAndDeregister("enfant");
logger.arriveAndDeregister("enfant");

00:00:00.091 [main           ] {racine    } création
00:00:00.141 [main           ] {racine    } phase=00 parties=00
00:00:00.142 [main           ] 
00:00:00.142 [main           ] {racine    } register
00:00:00.142 [main           ] {racine    } phase=00 parties=01
00:00:00.142 [main           ] 
00:00:00.143 [main           ] {enfant    } création
00:00:00.143 [main           ] {racine    } phase=00 parties=01
00:00:00.143 [main           ] {enfant    } phase=00 parties=00
00:00:00.144 [main           ] 
00:00:00.144 [main           ] {enfant    } register
00:00:00.144 [main           ] {racine    } phase=00 parties=02
00:00:00.145 [main           ] {enfant    } phase=00 parties=01
00:00:00.145 [main           ] 
00:00:00.145 [main           ] {enfant    } register
00:00:00.145 [main           ] {racine    } phase=00 parties=02
00:00:00.146 [main           ] {enfant    } phase=00 parties=02
00:00:00.146 [main           ] 
00:00:00.146 [main           ] {enfant    } arriveAndDeregister
00:00:00.147 [main           ] {racine    } phase=00 parties=02
00:00:00.147 [main           ] {enfant    } phase=00 parties=01
00:00:00.147 [main           ] 
00:00:00.147 [main           ] {enfant    } arriveAndDeregister
00:00:00.148 [main           ] {racine    } phase=00 parties=01
00:00:00.148 [main           ] {enfant    } phase=00 parties=00

Un Exchanger est un élément de synchronisation simpliste qui permet à deux threads d'échanger une donnée :

//com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.exchanger.ExchangerDemo
Exchanger<String> exchanger = new Exchanger<>();
Runnable tache = new Runnable() {
  public void run() {
    String envoi = Thread.currentThread().getName();
    Logger.println("Envoi %s", envoi);
    String recu = exchanger.exchange(envoi);
    Logger.println("Reçu  %s", recu);
  }
};
new Thread(tache, "Foo").start();
new Thread(tache, "Bar").start();

00:00:00.020 [Bar            ] Envoi Bar
00:00:00.020 [Foo            ] Envoi Foo
00:00:00.028 [Foo            ] Reçu  Bar
00:00:00.028 [Bar            ] Reçu  Foo

Les Exchangers sont particulièrement utiles dans le cas de Producteur-Consommateur ayant besoin d'échanger des informations.

Tous les exemples donnés dans cet article sont disponibles sous la forme d'un projet Maven hébergé sous GitHub. Tous les exemples cités contiennent une première ligne commentaire indiquant l'emplacement du fichier dans les sources. Les sources propres à ce chapitre se trouve sous le package com.developpez.lmauzaize.java.concurrence.ch02_synchronisation.

Si vous ne savez pas comment importer le projet, je vous invite à consulter l'article « Importer un projet Maven dans Eclipse en 5 minutes ».

Java Concurrent Animated est un projet Swing visant à montrer graphiquement le comportement de différents composants de l'API concurrente de Java.