Tutoriel sur la programmation concurrente en Java

Partie 2 : Éléments de synchronisation (natifs et API)

Logan Mauzaize

La programmation concurrente est un enjeu important et parfois difficile pour les développeurs. Cette série d'articles vise à vous présenter les différentes API disponibles en standard avec Java.

Les articles de la série :

- Gestion des threads

- Éléments de synchronisation (natifs et API)

- Structures de données

- Verrous

- Variables sans verrou

- Conseils et astuces

Ce second article vise à présenter les éléments natifs (propre au langage) et ceux de l'API qui permettent de définir des zones critiques dans votre code.

Commentez

- Titre : Tutoriel sur la programmation concurrente en Java

- Auteur : Logan Mauzaize

- Parution : 13 juin 2016

- Public visé :

- Licence : Les sources présentées sur cette page sont libres de droits et vous pouvez les utiliser à votre convenance. Par contre, la page de présentation constitue une œuvre intellectuelle protégée par les droits d'auteur. Copyright ® 2016 Logan Mauzaize. Aucune reproduction, même partielle, ne peut être faite de ce site et de l'ensemble de son contenu : textes, documents, images, etc. sans l'autorisation expresse de l'auteur. Sinon vous encourez selon la loi jusqu'à trois ans de prison et jusqu'à 300 000 € de dommages et intérêts.

Sommaire

- I. Présentation

- II. Éléments natifs

- II-A. Synchronisation

- II-B. volatile

- III. Éléments de l'API

- III-A. Semaphore

- III-B. CountDownLatch

- III-C. CyclicBarrier

- III-D. Phaser (1.7+)

- III-E. Exchanger

- IV. Conclusion

- V. Remerciements

- VI. Annexes

- VI-A. Sources des exemples

- VI-B. Java Concurrent Animated

I - Présentation

Précédemment, nous avons vu comment exécuter plusieurs tâches en parallèle et gérer leur exécution. Néanmoins, il arrive très souvent que le code présente des zones critiques. Il faut alors limiter le nombre de thread pouvant exécuter simultanément une portion précise de code. C'est ce que permettent les éléments de synchronisation.

II - Éléments natifs

II-A - Synchronisation

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 :

bloc synchronized (code)

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

bloc synchronized (console)

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) :

Thread.holdsLock(Object) (code)

//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));

Thread.holdsLock(Object) (console)

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

Il est donc bien important de noter que c'est l'instance qui est protégée et non la variable ! De fait, prenez garde à ne pas utiliser les blocs synchronized avec une valeur null.

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 :

IllegalMonitorStateException (code)

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

IllegalMonitorStateException (console)

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)

//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)

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)

//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)

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)

//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)

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 :

Échange de données avec synchronized (code)

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

Échange de données avec synchronized (console)

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

Il est important de noter ici la présence d'une boucle autour de l'appel à la méthode wait. En effet, il est possible de manière exceptionnelle de sortir de l'attente sans qu'aucune des trois conditions (timeout, notify, interruption) n'ait été remplie. On parle alors de « réveil fallacieux » (spurious wake-up), il convient alors de placer l'appel dans une boucle qui vérifiera que la condition attendue n'est toujours pas atteinte. Ce comportement est décrit dans la documentation de la méthode wait.

II-B - volatile

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

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

III - Éléments de l'API

synchronized et volatile offrent des services simples, mais parfois trop simples pour représenter des problèmes de synchronisation complexes. C'est pourquoi le package java.util.concurrent a été créé offrant ainsi différentes classes utilitaires pour vos différents besoins en synchronisation.

III-A - Semaphore

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.

Semaphore - démo (code)

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

Semaphore - démo (console)

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

Contrairement aux verrous (ex. : synchronized), les permissions ne sont attachées à aucun thread. Ainsi il n'y a pas de nécessité à effectuer la libération sur le même thread. Cependant, cela implique également qu'il ne gère pas la « réentrance ». Réaliser une acquisition supplémentaire dans un même thread revient à demander des permissions supplémentaires.

Semaphore - deadlock (code)

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

Semaphore - deadlock (console)

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.

Semaphore - équité

//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);

Vous pouvez jouer avec la variable équité pour changer l'équité du Semaphore. Néanmoins d'après mes tests sur la VM Oracle (1.7.0_45, 1.8.0_45), l'ordre de mise en attente est toujours respecté.

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

III-B - CountDownLatch

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

CountDownLatch - démo (code)

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

CountDownLatch - démo (console)

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)

//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)

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)

//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)

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

Il est important de noter qu'un CountDownLatch est jetable (« one-shot »). Une fois que le compteur a atteint zéro, il ne peut plus être réutilisé pour mettre des tâches en attente. Si vous avez besoin d'une version qui se réinitialise, il faut utiliser une CyclicBarrier.

III-C - CyclicBarrier

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.

CyclicBarrier - démo (code)

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

CyclicBarrier - démo (console)

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)

//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)

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)

//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)

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)

//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)

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

On constatera que dans le dernier cas, la barrière est prête à être réutilisée par les threads. Dans les autres cas, il faudra qu'un thread (un de ceux en attente ou un autre) réinitialise la barrière pour qu'elle soit de nouveau opérationnelle. Dans ce cas, il sera certainement nécessaire d'utiliser un autre mécanisme de synchronisation pour s'assurer qu'un seul thread réalise la réinitialisation.

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.

CyclicBarrier - action (code)

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

CyclicBarrier - action (console)

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

III-D - Phaser (1.7+)

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

Phaser - démo (code)

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

Phaser - démo (console)

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 :

Phaser - étapes (code)

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

Phaser - étapes (console)

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

Cet exemple illustre également comment coder l'atteinte d'une phase particulière.

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 :

Phaser - terminaison (code)

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

Phaser - terminaison (console)

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.

Phaser - niveaux (code)

//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");

Phaser - niveaux (console)

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

III-E - Exchanger

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

Exchanger - démo (code)

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

Exchanger - démo (console)

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.

IV - Conclusion

Au cours de ce second article, nous avons vu comment protéger les zones critiques de notre code. Cependant les éléments de synchronisation sont des éléments simples et nécessiteront souvent d'être utilisés dans des abstractions plus grandes. Ainsi le prochain article sera dédié aux structures de données et les implémentations dédiées à la concurrence.

V - Remerciements

Je remercie tous les contributeurs de l'API Java et les packages java.util.concurrent.* et plus particulièrement Doug Lea qui est l'un des principaux auteurs.

Je remercie également Mickael Baron, Thierry Leriche-Dessirier alias thierryler et Claude Leloup pour leur relecture attentive, leurs remarques et leurs bons conseils.

Je tiens aussi à remercier la communauté Developpez.com qui a mis en place tous les outils, procédures et l'hébergement nécessaires à la publication de cet article.

Enfin mon épouse et mes enfants pour leur patience et leur tolérance durant les nombreuses heures qui ont été nécessaires à la rédaction de cet article.

VI - Annexes

VI-A - Sources des exemples

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

VI-B - Java Concurrent Animated

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