.
I. Introduction▲
Un processus représente l'environnement d'exécution d'un programme. Il référence d'une part un espace mémoire permettant de stocker les données propres à l'application, et d'autre part un ensemble de threads permettant l'exécution du code qui manipulera ces données.
En Java, au démarrage de l'application, un thread initial est créé : le thread "main". Son rôle est de localiser le point d'entrée de l'application (la méthode public static void main(String... args)) puis d'exécuter son code.
Ce thread, comme tous les threads, exécute la séquence d'instructions qui lui est confiée de manière purement séquentielle. Si une instruction prend du temps à compléter (par exemple, en attente de connexion à un serveur), toute l'application est paralysée.
Pour éviter cela, il est possible (et même souhaitable) de confier l'exécution de ces portions bloquantes à des threads annexes, laissant ainsi le thread principal libre de continuer l'exécution de l'application.
Voyons comment.
II. La classe java.lang.Thread▲
II-A. Créer et démarrer un thread▲
En Java, un thread est représenté par une instance de la classe java.lang.Thread. Le code qu'il doit exécuter est défini dans sa méthode run(), et un simple appel à la méthode start() permet de le démarrer.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
public class NewThread {
public static void main(String... args) {
Thread t = new Thread() {
public void run() {
System.out.println("Je suis dans le thread : " + Thread.currentThread().getName()); // #1
}
};
t.start();
System.out.println("Je suis dans le thread : " + Thread.currentThread().getName()); // #2
}
}
Ce programme produit le résultat suivant :
2.
Je suis dans le thread : main
Je suis dans le thread : Thread-0
On voit ici que la ligne #2 a été exécutée dans le thread principal de l'application, alors que la ligne #1 a été exécutée dans un autre thread ("Thread-0").
II-B. Nommer les threads▲
Dans l'exemple ci-dessus, nous reconnaissons le thread "main", thread initial de l'application. Nous constatons également que notre thread annexe s'appelle "Thread-0".
Ce nom lui a été attribué automatiquement par la JVM, mais n'est pas très parlant. Pour faciliter le débogage, il est recommandé de donner des noms explicites à nos threads, en les passant au constructeur :
Thread t = new Thread("Mon thread") { ... };
Cette modification dans notre exemple produirait alors le résultat suivant :
2.
Je suis dans le thread : main
Je suis dans le thread : Mon thread
II-C. Start vs Run▲
Il est facile de confondre les méthodes run() et start() ; une erreur classique consiste à essayer de démarrer un nouveau thread en appelant la première au lieu de la seconde. Dans ce cas, le code de la méthode run() est bien exécuté, mais comme un simple appel de méthode dans le thread courant : aucun nouveau thread n'est lancé !
Il est facile de le vérifier :
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
public class NewThread {
public static void main(String... args) {
Thread t = new Thread() {
public void run() {
System.out.println("Je suis dans le thread : " + Thread.currentThread().getName());
}
};
t.run(); // ERREUR !
System.out.println("Je suis dans le thread : " + Thread.currentThread().getName());
}
}
Nous constatons que les deux instructions println() ont été exécutées dans le même thread :
2.
Je suis dans le thread : main
Je suis dans le thread : main
Pensez donc bien à appeler start() et non pas run() lorsque vous souhaitez démarrer un thread !
II-D. Vie et mort d'un thread▲
Une fois lancé, un thread "vit" jusqu'à ce qu'il ait fini d'exécuter le code de sa méthode run(). Après quoi, il est considéré comme "mort" ("terminated") et ne peut plus être redémarré.
Notez que la méthode stop() présente dans son API ne doit jamais être utilisée, car elle pourrait avoir des conséquences terribles et imprévisibles (!) sur lesquelles je ne m'étendrai pas ici.
III. Runnable▲
L'utilisation que nous faisons de la classe java.lang.Thread ci-dessus a le défaut de lier fortement la définition du traitement à exécuter (le "quoi") au thread particulier qui l'exécute (le "comment"). Que faire si l'on souhaite faire exécuter le même traitement par plusieurs threads ? Ou relancer un traitement après la mort du thread associé ? Nous comprenons ici la nécessité de découpler le "quoi" du "comment".
C'est ici qu'intervient l'interface java.lang.Runnable, qui permet d'encapsuler un traitement sous la forme d'un composant autonome et réutilisable. Pour être réellement exécuté, un Runnable doit être passé en paramètre à un Thread ou un ExecutorService (voir plus loin).
Dans l'exemple suivant, un même Runnable est passé à deux threads :
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
public class NewThreadWithRunnable {
public static void main(String... args) {
// Notre traitement, encapsulé dans un Runnable
Runnable job = new Runnable() { // #1
public void run() {
System.out.println("Je suis dans le thread : " + Thread.currentThread().getName());
}
};
Thread t1 = new Thread(job, "Premier thread"); //#2
t1.start();
Thread t2 = new Thread(job, "Second thread");
t2.start();
System.out.println("Je suis dans le thread : " + Thread.currentThread().getName());
}
}
En #2 nous passons le traitement à exécuter, encapsulé dans un Runnable en #1, à deux threads.
2.
3.
Je suis dans le thread : Premier thread
Je suis dans le thread : main
Je suis dans le thread : Second thread
Notez que l'ordre des lignes peut varier chez vous, car l'ordre dans lequel le processeur choisit d'exécuter les différents threads est imprévisible.
IV. Le framework Executor▲
Jusqu'ici, nous avons créé et lancé manuellement un nouveau Thread à chaque fois que nous souhaitions exécuter un traitement de manière asynchrone. Si cette technique fonctionne bien pour les petites applications, elle est fortement déconseillée pour les applications d'entreprise :
- d'une part, chaque thread supplémentaire augmente la mémoire consommée, la complexité globale de l'application, et le risque de contention ;
- d'autre part, pour de petites tâches, le coût de création d'un nouveau thread peut se révéler supérieur au coût d'exécution du traitement associé.
Introduit avec Java 5, le framework Executor répond à ces problématiques.
Disponible dans le package java.util.concurrent, il fournit un pool de threads robuste et hautement configurable, ainsi que les classes Callable et Future qui étendent les fonctionnalités des Runnables (voir plus loin). Sa mise en œuvre doit systématiquement être préférée à la création manuelle de threads.
IV-A. Executor et ExecutorService▲
L'interface Executor capture l'essence même du framework Executor : l'exécution d'un traitement encapsulé dans un Runnable.
2.
3.
public interface Executor {
void execute(Runnable command);
}
Je vous sens un peu déçus. Effectivement, c'est assez pauvre.
En réalité, toute la puissance du framework est exprimée dans une autre interface : ExecutorService, qui dérive d'Executor. Elle fournit des méthodes pour soumettre des traitements à exécuter (individuels ou en masse), ainsi que pour gérer le cycle de vie du pool :
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
public interface ExecutorService extends Executor {
// Job submission
public void submit(Runnable job);
public Future<V> submit(Callable<V> job);
// Lifecycle management
public void shutdown();
public void shutdownNow();
public boolean isShutdown();
// (...)
}
La classe ThreadPoolExecutor, qui implémente cette interface, est tellement configurable que ses auteurs ont préféré fournir une factory couvrant les besoins les plus courants :
2.
3.
4.
5.
6.
7.
8.
public class Executors {
public ExecutorService newSingleThreadExecutor() {...};
public ExecutorService newFixedThreadPool(int nbThreads) {...};
public ExecutorService newCachedThreadPool() {...};
public ExecutorService newSingleThreadScheduledExecutor() {...};
public ExecutorService newScheduledThreadPool(int nbThreads) {...};
// (...)
}
Voyons comment utiliser tout ceci pour exécuter nos traitements :
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
public class NewExecutorService {
public static void main(String... args) {
Runnable job = new Runnable() {
public void run() {
System.out.println("Je suis dans le thread : " + Thread.currentThread().getName());
}
};
// Pool avec 4 threads
ExecutorService pool = Executors.newFixedThreadPool(4);
pool.submit(job);
pool.submit(job);
pool.shutdown();
System.out.println("Je suis dans le thread : " + Thread.currentThread().getName());
}
}
Le résultat est le suivant :
2.
3.
Je suis dans le thread : pool-1-thread-1
Je suis dans le thread : pool-1-thread-2
Je suis dans le thread : main
Afin de prouver que le pool recycle les threads au lieu d'en recréer, tentons de soumettre notre Runnable dix fois :
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
Je suis dans le thread : pool-1-thread-1
Je suis dans le thread : pool-1-thread-4
Je suis dans le thread : pool-1-thread-3
Je suis dans le thread : pool-1-thread-2
Je suis dans le thread : pool-1-thread-2
Je suis dans le thread : pool-1-thread-2
Je suis dans le thread : pool-1-thread-2
Je suis dans le thread : pool-1-thread-3
Je suis dans le thread : pool-1-thread-4
Je suis dans le thread : pool-1-thread-1
Je suis dans le thread : main
Nous voyons ici que le thread "pool-1-thread-1" a été sollicité deux fois, le thread "pool-1-thread-2" quatre fois, etc.
IV-B. Callable et Future▲
IV-B-1. Callable▲
Nous avons vu plus haut comment encapsuler un traitement dans un Runnable. Mais Runnable montre vite ses limites : comment renvoyer un résultat ? Et si le traitement lève une exception ?
Pour lever ces limitations, le framework Executor propose l'interface Callable<V>, qui est une sorte de Runnable amélioré. Le type paramétré <V> définit le type du résultat produit par la méthode call(). Ainsi, un Callable<Integer> produira un Integer.
2.
3.
public interface Callable<V> {
V call() throws Exception;
}
Un Callable est prévu pour être soumis à la méthode submit() d'un pool de threads, qui exécutera son traitement de manière asynchrone.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
public class NewExecutorServiceWithCallable {
public static void main(String[] args) {
Callable<Void> job = new Callable<Void>() {
public Void call() throws Exception{
System.out.println("Je suis dans le thread " + Thread.currentThread().getName());
return null;
}
};
ExecutorService pool = Executors.newFixedThreadPool(4);
pool.submit(job);
pool.shutdown();
System.out.println("Je suis dans le thread " + Thread.currentThread().getName());
}
}
Le résultat est le suivant :
2.
Je suis dans le thread : pool-1-thread-1
Je suis dans le thread : main
IV-B-2. Future▲
Une fois soumis à un pool de threads, un Callable est généralement[1] exécuté de manière asynchrone ; le résultat produit ne sera sans doute pas disponible avant un certain temps.
Du point de vue de l'appelant (celui qui soumet le traitement au pool), cela n'aurait aucun sens d'attendre ce résultat de manière synchrone : il perdrait tout le bénéfice du système ! Mais il lui faut tout de même un moyen de récupérer le résultat lorsqu'il aura été calculé.
Le framework Executor fournit la classe Future<V>, qui représente un résultat de type V dont la valeur est pour l'instant inconnue (puisque le traitement n'a pas encore été exécuté), mais qui sera disponible dans le futur.
L'interface Future<V> propose un ensemble de méthodes permettant de récupérer le résultat (get()), tester sa disponibilité (isDone()), ou d'annuler son calcul (cancel()).
2.
3.
4.
5.
6.
7.
public interface Future<V> {
public V get();
public V get(long timeout, TimeUnit unit);
public boolean isDone();
public boolean cancel(boolean mayInterruptIfRunning);
public boolean isCancelled();
}
La méthode get() permet de récupérer le résultat immédiatement s'il est disponible (c'est-à-dire si le traitement a bien été exécuté par le pool de threads). Mais attention : si son calcul n'est pas terminé, la méthode est bloquante ! C'est donc une bonne pratique de vérifier la réelle disponibilité du résultat avec isDone() avant de le récupérer.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
public class FutureDemo {
public static void main(String[] args) throws InterruptedException, ExecutionException {
ExecutorService pool = Executors.newFixedThreadPool(4);
Future<Integer> result = pool.submit(new DeepThoughtCalculator()); // Should take 7.5s
pool.shutdown();
long timeBefore = System.currentTimeMillis();
while (!result.isDone()) {
// Do something useful
System.out.printf("Still nothing after %d ms, waiting a bit more... %n", System.currentTimeMillis() - timeBefore);
// Wait a bit and retry
Thread.sleep(1000);
}
Integer answer = result.get();
System.out.printf("Result after %dms : %d %n", System.currentTimeMillis()-timeBefore, answer);
}
}
Évidemment, dans une véritable application, on profitera de l'indisponibilité du résultat pour réaliser d'autres opérations utiles à l'application.
2.
3.
4.
5.
6.
7.
8.
9.
Still nothing after 0 ms, waiting a bit more...
Still nothing after 1014 ms, waiting a bit more...
Still nothing after 2015 ms, waiting a bit more...
Still nothing after 3015 ms, waiting a bit more...
Still nothing after 4015 ms, waiting a bit more...
Still nothing after 5016 ms, waiting a bit more...
Still nothing after 6016 ms, waiting a bit more...
Still nothing after 7016 ms, waiting a bit more...
Result after 8016ms : 42
IV-C. ExecutorService en action▲
Je suis sûr que vous brûlez de connaître la réponse à "La grande question sur la vie, l'univers et le reste" ?
Deux solutions s'offrent à nous :
- la demander directement à Deep Thought - mais le calcul risque de prendre 7.5 millions d'années ;
- la calculer selon la méthode des anciens Babyloniens, en espérant qu'elle soit plus rapide.
Encapsulons ces deux traitements dans des Callable :
2.
3.
4.
5.
6.
7.
8.
9.
public class DeepThoughtCalculator implements Callable<Integer> {
@Override
public Integer call() throws Exception {
Util.busyWait(7500, TimeUnit.MILLISECONDS); // Busy wait
return 42;
}
}
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
public class BabylonianCalculator implements Callable<Integer> {
private static final int MAGIC_NUMBER = 1764;
@Override
public Integer call() throws Exception {
int number = MAGIC_NUMBER;
double approximate = getApproximate(number);
while (!isPreciseEnough(number, approximate)) {
approximate = 0.5 * (approximate + (number / approximate));
Util.busyWait(250, TimeUnit.MILLISECONDS);
}
return (int) approximate;
}
private double getApproximate(int number) {
int digits = Integer.toString(number).length();
double approximate = Math.pow(10, digits);
return approximate *= number % 2 == 0 ? 2 : 6;
}
private boolean isPreciseEnough(int number, double approximate) {
return Math.abs(number - (approximate * approximate)) < 1;
}
}
Chacun de ces traitements est coûteux : 7.5 secondes pour le premier, environ 3 secondes (sur ma machine) pour le second dans notre exemple.
Essayons de les lancer séquentiellement :
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
public class SingleThreadedTest {
@Test
public void findAnswer() throws Exception {
long timeBefore = System.currentTimeMillis();
// Asking the ancient Babylonian science for an answer
int babylonianAnswer = new BabylonianCalculator().call();
System.out.printf("Babylonian result after %dms : %d %n", System.currentTimeMillis()-timeBefore, babylonianAnswer);
// Asking a big computer for an answer
int deepThoughtAnswer = new DeepThoughtCalculator().call();
System.out.printf("DeepThought result after %dms : %d %n", System.currentTimeMillis()-timeBefore, deepThoughtAnswer);
System.out.println((System.currentTimeMillis() - timeBefore) + " ms");
}
}
2.
3.
Babylonian result after 3002ms : 42
DeepThought result after 10503ms : 42
10503 ms
Leur exécution séquentielle prend environ 10.5 secondes et n'utilise qu'un seul processeur :
Essayons maintenant de les soumettre à un ExecutorService :
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
public class MultithreadedTest {
@Test
public void findAnswer() throws ExecutionException, InterruptedException {
long timeBefore = System.currentTimeMillis();
Callable<Integer> babylonianQuestion = new BabylonianCalculator();
Callable<Integer> deepThoughtQuestion = new DeepThoughtCalculator();
// Pool with 4 threads
ExecutorService pool = Executors.newFixedThreadPool(4);
Future<Integer> babylonianFuture = pool.submit(babylonianQuestion);
Future<Integer> deepThoughtFuture = pool.submit(deepThoughtQuestion);
pool.shutdown();
// Get result 1
Integer babylonianAnswer = babylonianFuture.get();
System.out.printf("Babylonian result after %dms : %d %n", System.currentTimeMillis()-timeBefore, babylonianAnswer);
// Get result 2
Integer deepThoughtAnswer = deepThoughtFuture.get();
System.out.printf("DeepThought result after %dms : %d %n", System.currentTimeMillis()-timeBefore, deepThoughtAnswer);
long timeAfter = System.currentTimeMillis();
System.out.println((timeAfter - timeBefore) + " ms");
}
}
2.
3.
Babylonian result after 3004ms : 42
DeepThought result after 7504ms : 42
7504 ms
Grâce au pool de threads, les deux calculs ont été menés en parallèle. Le temps d'exécution global a été réduit, et deux processeurs ont été utilisés :
Un problème se pose toutefois : l'ordre dans lequel les résultats ont été récupérés (en appelantget()) est important. Ici, nous avons - par chance! - récupéré en premier le résultat le plus rapidement disponible. Si nous avions inversé l'ordre des get(), nous aurions obtenu :
2.
3.
DeepThought result after 7506ms : 42
Babylonian result after 7506ms : 42
7506 ms
Bien que disponible au bout de 3 secondes seulement, le résultat calculé par la méthode babylonienne n'a pu être récupéré qu'au bout de 7.5 secondes !
IV-D. CompletionService▲
Ce problème d'ordre de disponibilité des résultats est fréquent. Pour le résoudre, le framework Executor propose la classe CompletionService.
CompletionService est un wrapper qui se branche sur un ExecutorService, et qui se charge de surveiller l'état d'avancement des différents traitements qui lui ont été soumis. Sa méthode take() (bloquante) renvoie les résultats au fur et à mesure de leur disponibilité.
Notez qu'il est tout de même nécessaire de connaître le nombre de résultats attendus.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
public class CompletionServiceDemo {
@Test
public void findAnswer() throws ExecutionException, InterruptedException {
// Wait for VM warmup
Thread.sleep(2000);
long timeBefore = System.currentTimeMillis();
Callable<Integer> babylonianQuestion = new BabylonianCalculator();
Callable<Integer> deepThoughtQuestion = new DeepThoughtCalculator();
// Pool with 4 threads
ExecutorService pool = Executors.newFixedThreadPool(4);
CompletionService<Integer> completion = new ExecutorCompletionService<Integer>(pool);
completion.submit(babylonianQuestion);
completion.submit(deepThoughtQuestion);
pool.shutdown();
Future<Integer> answer1 = completion.take();
System.out.printf("Result 1 after %dms : %d %n", System.currentTimeMillis()-timeBefore, answer1.get());
Future<Integer> answer2 = completion.take();
System.out.printf("Result 2 after %d ms : %d %n", System.currentTimeMillis()-timeBefore, answer2.get());
long timeAfter = System.currentTimeMillis();
System.out.println((timeAfter - timeBefore) + " ms");
}
}
2.
3.
Result 1 after 3005ms : 42
Result 2 after 7505 ms : 42
7505 ms
IV-E. InvokeAny : que le meilleur gagne !▲
Pour finir, voyons une fonctionnalité moins connue de l'ExecutorService : la méthode invokeAny().
Il arrive que plusieurs traitements (algorithmes ou sous-systèmes de l'application) soient en compétition pour produire un résultat donné - par exemple, récupérer une bibliothèque à partir de l'un des nombreux repositories Maven distants. Nous souhaiterions alors pouvoir tout arrêter dès que l'un des traitements renvoie le résultat souhaité.
La méthode invokeAny() prend en paramètre une collection de traitements de même type, et renvoie le résultat fourni par le traitement le plus rapide. Attention toutefois, invokeAny() est une méthode bloquante.
Dans notre exemple, deux algorithmes sont en compétition pour calculer la réponse à "La grande question sur la vie, l'univers et le reste", mais une seule réponse nous suffit. Que le meilleur gagne !
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
public class MultithreadedTest2 {
@Test
public void findAnswer() throws ExecutionException, InterruptedException {
// Wait for VM warmup
Thread.sleep(1000);
long timeBefore = System.currentTimeMillis();
Callable<Integer> deepThoughtQuestion = new DeepThoughtCalculator();
Callable<Integer> babylonianQuestion = new BabylonianCalculator();
// Submit many jobs, but waiting for only 1 answer
ExecutorService pool = Executors.newFixedThreadPool(4);
Integer answer = pool.invokeAny(Arrays.asList(deepThoughtQuestion, babylonianQuestion));
pool.shutdown();
System.out.printf("First result after %dms : %d %n", System.currentTimeMillis()-timeBefore, answer);
long timeAfter = System.currentTimeMillis();
System.out.println((timeAfter - timeBefore) + " ms");
}
}
2.
First result after 3004ms : 42
3004 ms
V. Conclusion▲
Dans cet article, nous avons vu les principales solutions offertes par Java pour exécuter différents traitements de manière concurrente, et tirer le meilleur parti de la puissance de calcul disponible sur la machine.
S'il est simple de créer et de lancer manuellement des threads, la solution recommandée est d'utiliser le framework Executor, plus souple, plus robuste, et disposant de nombreuses fonctionnalités.
Le prochain article décrira les problèmes qui peuvent survenir lorsque plusieurs threads tentent d'accéder à une même donnée.
Stay tuned for more happy days !
VI. Remerciements▲
Cet article a été publié avec l'aimable autorisation de la société ZenikaZenika, le billet original peut être trouvé sur le blog de ZenikaBlog de Zenika.
Nous tenons à remercier ClaudeLELOUP pour sa relecture attentive de cet article et Mickael Baron pour la mise au gabarit du billet original.