Tutoriel sur la programmation concurrente en Java

Partie 1 : Gestion des threads

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 premier article sert de préambule afin de poser les bases de la programmation parallèle en Java, à savoir la gestion des threads.

Commentez

- Titre : Tutoriel sur la programmation concurrente en Java

- Auteur : Logan Mauzaize

- Parution : 18 février 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 ® 2014 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. Modèle

- III. Gestion des threads

- III-A. ThreadFactory

- III-B. ExecutorService

- III-C. Future

- III-D. Fork/Join (1.7+)

- IV. Conclusion

- V. Remerciements

- VI. Annexes

- VI-A. Sources des exemples

- VI-B. Java Concurrent Animated

I - Présentation

Les ordinateurs ont vu leur nombre de processeurs/cœurs augmentés grandement ces dernières années afin de compenser les limites d'augmentation de puissance d'un seul processeur. Afin d'exploiter pleinement cette puissance de calcul, il convient de réaliser des applications capables d'effectuer plusieurs tâches en parallèle. Et même si les systèmes d'exploitation sont depuis longtemps multitâches, Java demeure un langage séquentiel à sa base.

Depuis longtemps, les développeurs utilisent deux techniques pour rendre leurs applications multitâches : le fork ou les threads. En Java, c'est cette dernière technique qui est utilisée. Contrairement au fork, les threads partagent la même mémoire.

Partageant ainsi des ressources (en mémoire), les threads risquent alors de rentrer en concurrence et de corrompre le système. C'est alors qu'intervient la programmation concurrente qui rassemble un ensemble de fonctionnalités et de techniques pour permettre la synchronisation de tâches s'exécutant en parallèle.

Cet article s'attache uniquement à la synchronisation des ressources en mémoire. En effet, les autres ressources étant partagées également par d'autres processus (éventuellement sur d'autres machines), les outils et techniques sont souvent propres à chaque ressource (base de données, disque, etc.).

Si ces ressources sont accédées uniquement par plusieurs threads du même programme Java, les outils et techniques décrits ci-après s'appliquent dans la mesure où vous créez un objet unique pour chacune de ces ressources.

II - Modèle

En Java, un thread est représenté par la classe java.lang.Thread. Les threads disposent de plusieurs propriétés :

- id : un numéro unique ;

- name : le nom du thread. Il n'est pas nécessairement unique ;

- priority : la machine virtuelle exécute prioritairement les threads avec la plus grande valeur ;

- state : l'état du thread ;

- threadGroup : permet d'organiser les threads ;

- daemon : indique si le thread est une tâche de fond (démon). La VM s'arrête automatiquement dès lors qu'aucun thread principal (non démon) n'est en cours d'exécution.

Les groupes de threads forment une arborescence. Lors de sa création, par défaut, un groupe appartient au groupe du thread en cours et hérite également de ses propriétés. Il en est de même pour les threads qui héritent des propriétés du thread en cours.

Pour exécuter du code dans un thread, il existe deux techniques :

1. Créer une classe fille de Thread et surcharger la méthode run() :

Logger

1. //com.developpez.lmauzaize.java.concurrence.Logger

2. public class Logger {

4. private static final long startTime = System.currentTimeMillis();

5. private static final TimeZone UTC = TimeZone.getTimeZone("UTC");

6. public static final UncaughtExceptionHandler exceptionHandler = new UncaughtExceptionHandler() {

7. @Override

8. public void uncaughtException(Thread t, Throwable e) {

9. printStackTrace(t, e);

10. }

11. public String toString() {

12. return "Logger.printStackTrace(Thread,Throwable)";

13. }

14. };

15. private Logger() {}

18. public static void println(String msg, Object... params) {

19. println(Thread.currentThread(), msg, params);

20. }

21. public static void println(Thread thread, String msg, Object... params) {

22. // Message

23. String format = "%tT.%<tL [%-15s] " + msg + "%n";

25. // Paramètres

26. Object[] args = new Object[2 + (params != null ? params.length : 0)];

27. Calendar cal = Calendar.getInstance(UTC);

28. cal.setTimeInMillis(System.currentTimeMillis() - startTime);

29. args[0] = cal;

30. args[1] = Thread.currentThread().getName();

31. if (args.length > 2) {

32. System.arraycopy(params, 0, args, 2, params.length);

33. }

34. System.out.printf(format, args);

35. }

37. public static void printStackTrace(Throwable t) {

38. printStackTrace(Thread.currentThread(), t);

39. }

40. public static void printStackTrace(Thread thread, Throwable t) {

41. StringWriter writer = new StringWriter();

42. t.printStackTrace(new PrintWriter(writer, true));

43. println(thread, "%s%n%s", t, writer);

44. }

45. }

La classe Logger sera utilisée tout au long de cet article pour faciliter l'affichage des traces sur la sortie standard. Vous noterez également que la première ligne est un commentaire indiquant le nom complet de la classe dans les sources de l'article.

Nouvelle classe fille de Thread

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.FilleThread

2. new Thread() {

3. public void run() {

4. Logger.println("Je suis un nouveau thread !");

5. }

6. }.start();

2. Créer une instance de Runnable et passer cette instance en paramètre lors de la création du thread. Cette dernière technique permet d'exécuter plusieurs fois le même code :

Nouvelle classe Runnable

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.NouveauRunnable

2. Runnable runnable = new Runnable() {

3. public void run() {

4. Logger.println("Je suis un nouveau thread !");

5. }

6. };

7. new Thread(runnable).start();

8. new Thread(runnable).start();

Certaines parties de l'API utilisent une classe similaire à Runnable qui permet à une tâche de renvoyer un résultat, il s'agit de l'interface Callable.

Pour terminer cette présentation des spécifications Java, il est important de noter que toute instance peut servir de verrou. Ce qui nous amène à présenter les blocs synchronized et les méthodes wait/notify.

III - Gestion des threads

La section Modèle présentait le fonctionnement de base des threads avec Java. Cette section vise à présenter les solutions proposées par l'API pour gérer l'exécution de tâches sans avoir à gérer manuellement les threads.

III-A - ThreadFactory

Bien que la gestion des threads puisse être déléguée à un service dédié (voir ci-après), leur création fait l'objet d'une interface particulière appelée ThreadFactory. Cette interface permet de spécifier les paramètres des threads à créer : groupe, nom, priorité, démon, etc.

L'API fournit deux méthodes pour créer des ThreadFactory :

- defaultThreadFactory : créer des threads avec les caractéristiques décrites ci-dessous :

- Nom : de la forme « pool-<N>-thread-<M> ». « N » étant un numéro séquentiel de fabrique et « M » un numéro séquentiel de thread au sein de la fabrique,

- Groupe : le groupe est déterminé à la création de la fabrique. S'il y a un gestionnaire de sécurité, c'est son groupe qui est utilisé ; sinon c'est celui du thread courant,

- Démon : la fabrique ne produit que des threads non démons,

- Priorité : la priorité est normale sauf si le maximum toléré par le groupe est inférieur.

Executors.defaultThreadFactory - démo (code)

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.threadfactory.DefaultThreadFactoryDemo

2. public static Thread créerThreadLimité(Runnable tâche, String name) {

3. ThreadGroup groupe = new ThreadGroup("groupe");

4. groupe.setDaemon(true);

5. groupe.setMaxPriority(Thread.MIN_PRIORITY);

6. Thread thread = new Thread(groupe, tâche, name);

7. thread.setDaemon(true);

8. class FauxClassLoader extends ClassLoader {}

9. thread.setContextClassLoader(new FauxClassLoader());

10. thread.setUncaughtExceptionHandler(Logger.exceptionHandler);

11. return thread;

12. }

14. public static void affiche(String cas, Thread) {

15. Logger.println("");

16. Logger.println(cas);

17. Logger.println("  nom     =%s", thread.getName());

18. Logger.println("  groupe  =%s", thread.getThreadGroup().getName());

19. Logger.println("  démon   =%s", thread.isDaemon());

20. Logger.println("  priorité=%s", thread.getPriority());

21. Logger.println("  loader  =%s", thread.getContextClassLoader());

22. Logger.println("  handler =%s", thread.getUncaughtExceptionHandler());

23. }

25. public static void main(String[] args) throws InterruptedException {

27. ThreadFactory defaultThreadFactory = Executors.defaultThreadFactory();

28. ThreadFactory customThreadFactory  = Thread::new;

30. Runnable tâche = () -> {

31. affiche("defaultThreadFactory", defaultThreadFactory.newThread(null));

32. affiche("customThreadFactory" , customThreadFactory.newThread(null));

33. };

35. // Depuis "main"

36. tâche.run();

38. // Depuis un nouveau thread

39. Thread defaultThread = defaultThreadFactory.newThread(tâche);

40. defaultThread.start();

41. defaultThread.join();

43. // Depuis un groupe restreint

44. Thread thread = créerThreadLimité(tâche, "thread");

45. thread.start();

46. thread.join();

47. }

Executors.defaultThreadFactory - démo (console)

1. 00:00:00.022 [main           ]

2. 00:00:00.031 [main           ]

3. 00:00:00.031 [main           ] ******************************

4. 00:00:00.031 [main           ] Depuis main

5. 00:00:00.032 [main           ]

6. 00:00:00.032 [main           ] defaultThreadFactory

7. 00:00:00.032 [main           ]   nom     =pool-1-thread-1

8. 00:00:00.032 [main           ]   groupe  =main

9. 00:00:00.033 [main           ]   démon   =false

10. 00:00:00.033 [main           ]   priorité=5

11. 00:00:00.033 [main           ]   loader  =AppClassLoader

12. 00:00:00.033 [main           ]   handler =java.lang.ThreadGroup[name=main,maxpri=10]

13. 00:00:00.034 [main           ]

14. 00:00:00.034 [main           ] customThreadFactory

15. 00:00:00.034 [main           ]   nom     =Thread-0

16. 00:00:00.034 [main           ]   groupe  =main

17. 00:00:00.035 [main           ]   démon   =false

18. 00:00:00.035 [main           ]   priorité=5

19. 00:00:00.035 [main           ]   loader  =AppClassLoader

20. 00:00:00.035 [main           ]   handler =java.lang.ThreadGroup[name=main,maxpri=10]

21. 00:00:00.035 [main           ]

22. 00:00:00.036 [main           ]

23. 00:00:00.036 [main           ] ******************************

24. 00:00:00.036 [main           ] Depuis un nouveau thread

25. 00:00:00.037 [pool-1-thread-2]

26. 00:00:00.037 [pool-1-thread-2] defaultThreadFactory

27. 00:00:00.038 [pool-1-thread-2]   nom     =pool-1-thread-3

28. 00:00:00.038 [pool-1-thread-2]   groupe  =main

29. 00:00:00.038 [pool-1-thread-2]   démon   =false

30. 00:00:00.039 [pool-1-thread-2]   priorité=5

31. 00:00:00.039 [pool-1-thread-2]   loader  =AppClassLoader

32. 00:00:00.039 [pool-1-thread-2]   handler =java.lang.ThreadGroup[name=main,maxpri=10]

33. 00:00:00.039 [pool-1-thread-2]

34. 00:00:00.040 [pool-1-thread-2] customThreadFactory

35. 00:00:00.040 [pool-1-thread-2]   nom     =Thread-1

36. 00:00:00.040 [pool-1-thread-2]   groupe  =main

37. 00:00:00.040 [pool-1-thread-2]   démon   =false

38. 00:00:00.040 [pool-1-thread-2]   priorité=5

39. 00:00:00.041 [pool-1-thread-2]   loader  =AppClassLoader

40. 00:00:00.041 [pool-1-thread-2]   handler =java.lang.ThreadGroup[name=main,maxpri=10]

41. 00:00:00.041 [main           ]

42. 00:00:00.041 [main           ]

43. 00:00:00.042 [main           ] ******************************

44. 00:00:00.042 [main           ] Depuis un groupe restreint

45. 00:00:00.042 [thread         ]

46. 00:00:00.043 [thread         ] defaultThreadFactory

47. 00:00:00.043 [thread         ]   nom     =pool-1-thread-4

48. 00:00:00.043 [thread         ]   groupe  =main

49. 00:00:00.044 [thread         ]   démon   =false

50. 00:00:00.044 [thread         ]   priorité=5

51. 00:00:00.045 [thread         ]   loader  =FauxClassLoader

52. 00:00:00.045 [thread         ]   handler =java.lang.ThreadGroup[name=main,maxpri=10]

53. 00:00:00.045 [thread         ]

54. 00:00:00.046 [thread         ] customThreadFactory

55. 00:00:00.046 [thread         ]   nom     =Thread-2

56. 00:00:00.046 [thread         ]   groupe  =groupe

57. 00:00:00.047 [thread         ]   démon   =true

58. 00:00:00.047 [thread         ]   priorité=1

59. 00:00:00.047 [thread         ]   loader  =FauxClassLoader

60. 00:00:00.048 [thread         ]   handler =java.lang.ThreadGroup[name=groupe,maxpri=1]

- privilegedThreadFactory : même chose que précédemment avec le ClassLoader et les mêmes permissions que le thread ayant créé la fabrique.

Executors.privilegedThreadFactory - démo (code)

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.threadfactory.PrivilegedThreadFactoryDemo

2. // Vérifie si le thread possède les permissions

3. static boolean possèdePermission();

4. // Effectue une action avec les permissions

5. static <T> T avecPermission(PrivilegedAction<T> action);

7. public static void afficheAvecPermission(String cas) {

8. affiche(cas, Thread.currentThread());

9. Logger.println("  permis  =%s", possèdePermission());

10. }

12. class Executeur implements Runnable {

14. Map<String, ThreadFactory> threadFactories = new LinkedHashMap<>();

15. Executeur() {

16. threadFactories.put("Thread.new"             , Thread::new);

17. threadFactories.put("defaultThreadFactory"   , Executors.defaultThreadFactory());

18. threadFactories.put("privilegedThreadFactory", Executors.privilegedThreadFactory());

19. }

21. public void run() {

22. for (Entry<String, ThreadFactory> threadFactory : threadFactories.entrySet()) {

23. Thread thread = threadFactory.getValue().newThread(() -> afficheAvecPermission(threadFactory.getKey()));

24. thread.start();

25. thread.join();

26. }

27. }

28. }

30. Logger.println("");

31. Logger.println("");

32. Logger.println("******************************");

33. Logger.println("Sans permission");

34. new Executeur().run();

36. Logger.println("");

37. Logger.println("");

38. Logger.println("******************************");

39. Logger.println("Avec permission");

40. Executeur executeur = avecPermission(Executeur::new);

42. Thread thread = créerThreadLimité(executeur, "tâche");

43. thread.start();

44. thread.join();

Executors.privilegedThreadFactory - démo (console)

1. 00:00:00.020 [main           ]

2. 00:00:00.028 [main           ]

3. 00:00:00.028 [main           ] ******************************

4. 00:00:00.028 [main           ] Sans permission

5. 00:00:00.048 [Thread-0       ]

6. 00:00:00.048 [Thread-0       ] Thread.new

7. 00:00:00.049 [Thread-0       ]   nom     =Thread-0

8. 00:00:00.049 [Thread-0       ]   groupe  =main

9. 00:00:00.049 [Thread-0       ]   démon   =false

10. 00:00:00.049 [Thread-0       ]   priorité=5

11. 00:00:00.050 [Thread-0       ]   loader  =AppClassLoader

12. 00:00:00.050 [Thread-0       ]   handler =java.lang.ThreadGroup[name=main,maxpri=10]

13. 00:00:00.175 [Thread-0       ]   permis  =false

14. 00:00:00.177 [pool-1-thread-1]

15. 00:00:00.177 [pool-1-thread-1] defaultThreadFactory

16. 00:00:00.177 [pool-1-thread-1]   nom     =pool-1-thread-1

17. 00:00:00.178 [pool-1-thread-1]   groupe  =main

18. 00:00:00.178 [pool-1-thread-1]   démon   =false

19. 00:00:00.179 [pool-1-thread-1]   priorité=5

20. 00:00:00.179 [pool-1-thread-1]   loader  =AppClassLoader

21. 00:00:00.180 [pool-1-thread-1]   handler =java.lang.ThreadGroup[name=main,maxpri=10]

22. 00:00:00.180 [pool-1-thread-1]   permis  =false

23. 00:00:00.186 [pool-2-thread-1]

24. 00:00:00.187 [pool-2-thread-1] privilegedThreadFactory

25. 00:00:00.187 [pool-2-thread-1]   nom     =pool-2-thread-1

26. 00:00:00.187 [pool-2-thread-1]   groupe  =main

27. 00:00:00.187 [pool-2-thread-1]   démon   =false

28. 00:00:00.187 [pool-2-thread-1]   priorité=5

29. 00:00:00.188 [pool-2-thread-1]   loader  =AppClassLoader

30. 00:00:00.188 [pool-2-thread-1]   handler =java.lang.ThreadGroup[name=main,maxpri=10]

31. 00:00:00.188 [pool-2-thread-1]   permis  =false

32. 00:00:00.189 [main           ]

33. 00:00:00.189 [main           ]

34. 00:00:00.189 [main           ] ******************************

35. 00:00:00.189 [main           ] Avec permission

36. 00:00:00.191 [Thread-1       ]

37. 00:00:00.191 [Thread-1       ] Thread.new

38. 00:00:00.192 [Thread-1       ]   nom     =Thread-1

39. 00:00:00.192 [Thread-1       ]   groupe  =groupe

40. 00:00:00.192 [Thread-1       ]   démon   =true

41. 00:00:00.192 [Thread-1       ]   priorité=1

42. 00:00:00.193 [Thread-1       ]   loader  =FauxClassLoader

43. 00:00:00.193 [Thread-1       ]   handler =java.lang.ThreadGroup[name=groupe,maxpri=1]

44. 00:00:00.193 [Thread-1       ]   permis  =false

45. 00:00:00.194 [pool-3-thread-1]

46. 00:00:00.194 [pool-3-thread-1] defaultThreadFactory

47. 00:00:00.194 [pool-3-thread-1]   nom     =pool-3-thread-1

48. 00:00:00.194 [pool-3-thread-1]   groupe  =main

49. 00:00:00.194 [pool-3-thread-1]   démon   =false

50. 00:00:00.195 [pool-3-thread-1]   priorité=5

51. 00:00:00.195 [pool-3-thread-1]   loader  =FauxClassLoader

52. 00:00:00.195 [pool-3-thread-1]   handler =java.lang.ThreadGroup[name=main,maxpri=10]

53. 00:00:00.195 [pool-3-thread-1]   permis  =false

54. 00:00:00.196 [pool-4-thread-1]

55. 00:00:00.196 [pool-4-thread-1] privilegedThreadFactory

56. 00:00:00.196 [pool-4-thread-1]   nom     =pool-4-thread-1

57. 00:00:00.196 [pool-4-thread-1]   groupe  =main

58. 00:00:00.196 [pool-4-thread-1]   démon   =false

59. 00:00:00.197 [pool-4-thread-1]   priorité=5

60. 00:00:00.197 [pool-4-thread-1]   loader  =AppClassLoader

61. 00:00:00.197 [pool-4-thread-1]   handler =java.lang.ThreadGroup[name=main,maxpri=10]

62. 00:00:00.197 [pool-4-thread-1]   permis  =true

III-B - ExecutorService

L'interface Executor désigne un service capable de traiter des tâches (Runnable). Elle est étendue par l'interface ExecutorService qui permet de suivre l'évolution du traitement des tâches via des Future (voir ci-après).

Pour créer une instance d'ExecutorService, le plus simple est d'utiliser la classe utilitaire Executors :

- newCachedThreadPool : utilise un pool de threads qui grandit en fonction des besoins. Les threads créés sont conservés jusqu'à 60 secondes d'inactivité. Il est donc idéal pour exécuter plusieurs tâches courtes, mais pas pour exécuter un grand nombre de tâches en parallèle, en effet il risque de saturer le système par la création d'un trop grand nombre de threads.

ThreadFactory - logger

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.ThreadFactoryLogger

2. public class ThreadFactoryLogger implements ThreadFactory {

3. private ThreadFactory threadFactory = Executors.defaultThreadFactory();

5. protected void début() {

6. Logger.println("Début");

7. }

8. protected void fin() {

9. Logger.println("Fin");

10. }

12. public Thread newThread(Runnable r) {

13. class Enveloppe implements Runnable {

14. public void run() {

15. try {

16. début();

17. r.run();

18. } finally {

19. fin();

20. }

21. }

22. }

23. return threadFactory.newThread(new Enveloppe());

24. }

25. }

Executors.newCachedThreadPool - démo (code)

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.executors.NewCachedThreadPoolDemo

2. int parallele = 4;

4. ExecutorService executor = Executors.newCachedThreadPool(new ThreadFactoryLogger());

5. for (int i = 1; i <= parallele; i++) {

6. String nom = String.format("Tache-%02d", i);

7. Runnable tache = new Runnable() {

8. public void run() {

9. Logger.println("[%s] Début", nom);

10. int pause = ThreadLocalRandom.current().nextInt(2_000) + 200;

11. Logger.println("[%s] Pause %dms", nom, pause);

12. TimeUnit.MILLISECONDS.sleep(pause);

13. Logger.println("[%s] Fin", nom);

14. }

15. };

16. executor.submit(tache);

17. }

Executors.newCachedThreadPool - démo (console)

1. 00:00:00.004 [pool-1-thread-4] Début

2. 00:00:00.004 [pool-1-thread-2] Début

3. 00:00:00.004 [pool-1-thread-1] Début

4. 00:00:00.004 [pool-1-thread-3] Début

5. 00:00:00.005 [pool-1-thread-1] [Tache-01] Début

6. 00:00:00.005 [pool-1-thread-2] [Tache-02] Début

7. 00:00:00.004 [pool-1-thread-4] [Tache-04] Début

8. 00:00:00.005 [pool-1-thread-3] [Tache-03] Début

9. 00:00:00.006 [pool-1-thread-2] [Tache-02] Pause 1554ms

10. 00:00:00.006 [pool-1-thread-3] [Tache-03] Pause 1136ms

11. 00:00:00.006 [pool-1-thread-1] [Tache-01] Pause 882ms

12. 00:00:00.006 [pool-1-thread-4] [Tache-04] Pause 1847ms

13. 00:00:00.889 [pool-1-thread-1] [Tache-01] Fin

14. 00:00:01.143 [pool-1-thread-3] [Tache-03] Fin

15. 00:00:01.561 [pool-1-thread-2] [Tache-02] Fin

16. 00:00:01.855 [pool-1-thread-4] [Tache-04] Fin

17. 00:01:00.896 [pool-1-thread-1] Fin

18. 00:01:01.150 [pool-1-thread-3] Fin

19. 00:01:01.568 [pool-1-thread-2] Fin

20. 00:01:01.862 [pool-1-thread-4] Fin

- newFixedThreadPool : utilise un pool de threads de taille fixe. Les threads sont démarrés au fur et à mesure des besoins jusqu'à ce que le pool atteigne sa taille maximale. Les threads ainsi démarrés vivent tant que le service n'est pas arrêté. Si tous les threads sont occupés, les tâches sont mises en attente dans une file pour une exécution ultérieure.

Executors.newFixedThreadPool - démo (code)

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.executors.FixedThreadPoolDemo

2. int parallele = 4;

4. Logger.println("Lancement du pool");

5. ExecutorService executor = Executors.newFixedThreadPool(parallele, new ThreadFactoryLogger());

6. for (int i = 1; i <= parallele * 2; i++) {

7. String nom = String.format("Tache-%02d", i);

8. Logger.println("Soumission de la tâche %s", nom);

9. int nb = i;

10. Runnable tache = new Runnable() {

11. public void run() {

12. int pause = ThreadLocalRandom.current().nextInt(2_000) + 1_500 + (parallele*2-nb)*500;

13. Logger.println("[%s] Pause %dms", nom, pause);

14. TimeUnit.MILLISECONDS.sleep(pause);

15. Logger.println("[%s] Fin", nom);

16. }

17. };

18. executor.submit(tache);

19. TimeUnit.MILLISECONDS.sleep(500);

20. }

21. TimeUnit.SECONDS.sleep(12);

22. executor.shutdown();

23. }

Executors.newFixedThreadPool - démo (console)

1. 00:00:00.019 [main           ] Lancement du pool

2. 00:00:00.030 [main           ] Soumission de la tâche Tache-01

3. 00:00:00.032 [pool-1-thread-1] Début

4. 00:00:00.033 [pool-1-thread-1] [Tache-01] Pause 6048ms

5. 00:00:00.532 [main           ] Soumission de la tâche Tache-02

6. 00:00:00.533 [pool-1-thread-2] Début

7. 00:00:00.533 [pool-1-thread-2] [Tache-02] Pause 6337ms

8. 00:00:01.033 [main           ] Soumission de la tâche Tache-03

9. 00:00:01.034 [pool-1-thread-3] Début

10. 00:00:01.034 [pool-1-thread-3] [Tache-03] Pause 4725ms

11. 00:00:01.534 [main           ] Soumission de la tâche Tache-04

12. 00:00:01.535 [pool-1-thread-4] Début

13. 00:00:01.535 [pool-1-thread-4] [Tache-04] Pause 4172ms

14. 00:00:02.035 [main           ] Soumission de la tâche Tache-05

15. 00:00:02.535 [main           ] Soumission de la tâche Tache-06

16. 00:00:03.035 [main           ] Soumission de la tâche Tache-07

17. 00:00:03.535 [main           ] Soumission de la tâche Tache-08

18. 00:00:05.708 [pool-1-thread-4] [Tache-04] Fin

19. 00:00:05.708 [pool-1-thread-4] [Tache-05] Pause 4514ms

20. 00:00:05.760 [pool-1-thread-3] [Tache-03] Fin

21. 00:00:05.760 [pool-1-thread-3] [Tache-06] Pause 3512ms

22. 00:00:06.082 [pool-1-thread-1] [Tache-01] Fin

23. 00:00:06.082 [pool-1-thread-1] [Tache-07] Pause 2547ms

24. 00:00:06.871 [pool-1-thread-2] [Tache-02] Fin

25. 00:00:06.871 [pool-1-thread-2] [Tache-08] Pause 2579ms

26. 00:00:08.630 [pool-1-thread-1] [Tache-07] Fin

27. 00:00:09.273 [pool-1-thread-3] [Tache-06] Fin

28. 00:00:09.451 [pool-1-thread-2] [Tache-08] Fin

29. 00:00:10.223 [pool-1-thread-4] [Tache-05] Fin

30. 00:00:16.036 [pool-1-thread-2] Fin

31. 00:00:16.036 [pool-1-thread-1] Fin

32. 00:00:16.037 [pool-1-thread-3] Fin

33. 00:00:16.036 [pool-1-thread-4] Fin

- newSingleThreadExecutor : même chose que précédemment avec une taille de 1, excepté que le service ne peut être reconfiguré (par exemple en transtypant).

Executors.newSingleThreadExecutor - démo (code)

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.executors.SingleThreadPoolDemo

2. ExecutorService executor = Executors.newSingleThreadExecutor(new ThreadFactoryLogger());

4. Logger.println("Description du service:");

5. afficherMembres(executor.getClass());

7. Logger.println("");

8. Random random = new Random();

9. for (int i = 1; i <= 4; i++) {

10. String nom = String.format("Tache-%02d", i);

11. Runnable tache = new Runnable() {

12. public void run() {

13. int pause = random.nextInt(2_000);

14. Logger.println("[%s] Pause %dms", nom, pause);

15. TimeUnit.MILLISECONDS.sleep(pause);

16. Logger.println("[%s] Fin", nom);

17. }

18. };

19. executor.submit(tache);

20. }

21. executor.shutdown();

Executors.newSingleThreadExecutor - démo (console)

1. 00:00:00.020 [main           ] Description du service:

2. 00:00:00.030 [main           ] Executors$FinalizableDelegatedExecutorService

3. 00:00:00.077 [main           ]  --> void shutdown()

4. 00:00:00.077 [main           ]  --> void execute(Runnable)

5. 00:00:00.078 [main           ]  --> boolean awaitTermination(long,TimeUnit)

6. 00:00:00.078 [main           ]  --> List invokeAll(Collection)

7. 00:00:00.078 [main           ]  --> List invokeAll(Collection,long,TimeUnit)

8. 00:00:00.078 [main           ]  --> Object invokeAny(Collection)

9. 00:00:00.079 [main           ]  --> Object invokeAny(Collection,long,TimeUnit)

10. 00:00:00.079 [main           ]  --> boolean isShutdown()

11. 00:00:00.079 [main           ]  --> boolean isTerminated()

12. 00:00:00.080 [main           ]  --> List shutdownNow()

13. 00:00:00.080 [main           ]  --> Future submit(Runnable)

14. 00:00:00.080 [main           ]  --> Future submit(Callable)

15. 00:00:00.081 [main           ]  --> Future submit(Runnable,Object)

16. 00:00:00.081 [main           ]  --> void wait()

17. 00:00:00.081 [main           ]  --> void wait(long,int)

18. 00:00:00.082 [main           ]  --> void wait(long)

19. 00:00:00.082 [main           ]  --> boolean equals(Object)

20. 00:00:00.082 [main           ]  --> String toString()

21. 00:00:00.083 [main           ]  --> int hashCode()

22. 00:00:00.083 [main           ]  --> Class getClass()

23. 00:00:00.083 [main           ]  --> void notify()

24. 00:00:00.083 [main           ]  --> void notifyAll()

25. 00:00:00.084 [main           ]

26. 00:00:00.086 [pool-1-thread-1] Début

27. 00:00:00.086 [pool-1-thread-1] [Tache-01] Pause 1749ms

28. 00:00:01.835 [pool-1-thread-1] [Tache-01] Fin

29. 00:00:01.835 [pool-1-thread-1] [Tache-02] Pause 1358ms

30. 00:00:03.194 [pool-1-thread-1] [Tache-02] Fin

31. 00:00:03.194 [pool-1-thread-1] [Tache-03] Pause 1220ms

32. 00:00:04.415 [pool-1-thread-1] [Tache-03] Fin

33. 00:00:04.415 [pool-1-thread-1] [Tache-04] Pause 1730ms

34. 00:00:06.146 [pool-1-thread-1] [Tache-04] Fin

35. 00:00:06.146 [pool-1-thread-1] Fin

Sachez qu'il existe une interface fille de ExecutorService qui permet également la planification : ScheduledExecutorService. Ce type de service permet de retarder l'exécution d'une tâche et/ou de les exécuter de manière périodique.

III-C - Future

Un Future représente une tâche à exécuter et permet de récupérer son résultat via la méthode get() :

- il peut s'agit de la valeur renvoyée :

Future - Récupération du résultat (code)

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.future.FutureRécupérationRésultat

2. Callable<Thread> tâche = new Callable<Thread>() {

3. public Thread call() throws Exception {

4. Logger.println("Début");

5. Thread.sleep(2_000);

6. Logger.println("Fin");

7. return Thread.currentThread();

8. }

9. };

10. ExecutorService executor = Executors.newSingleThreadExecutor();

11. Future<Thread> future = executor.submit(tâche);

12. Thread thread = future.get();

13. Logger.println("La tâche a été exécutée sur %s", thread.getName());

14. executor.shutdown();

Future - Récupération du résultat (console)

1. 00:00:00.019 [pool-1-thread-1] Début

2. 00:00:02.027 [pool-1-thread-1] Fin

3. 00:00:02.027 [main           ] La tâche a été exécutée sur pool-1-thread-1

- ou, il peut s'agir de l'exception levée :

Future - Récupération de l'exception (code)

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.future.FutureRécupérationException

2. class MonException extends Exception {}

3. Callable<Thread> tâche = new Callable<Thread>() {

4. public Thread call() throws Exception {

5. Logger.println("Début");

6. Logger.println("Fin");

7. throw new MonException();

8. }

9. };

10. ExecutorService executor = Executors.newSingleThreadExecutor();

11. try {

12. Future<Thread> future = executor.submit(tâche);

13. Thread thread = future.get();

14. Logger.println("La tâche a été exécutée sur %s", thread);

15. } catch (ExecutionException e) {

16. Logger.printStackTrace(e);

17. } finally {

18. executor.shutdown();

19. }

Future - Récupération de l'exception (console)

1. 00:00:00.018 [pool-1-thread-1] Début

2. 00:00:00.026 [pool-1-thread-1] Fin

3. 00:00:00.027 [main           ] java.util.concurrent.ExecutionException: com.developpez.lmauzaize.java.concurrence.thread.future.FutureRécupérationException$1MonException

4. java.util.concurrent.ExecutionException: com.developpez.lmauzaize.java.concurrence.thread.future.FutureRécupérationException$1MonException

5. at java.util.concurrent.FutureTask.report(FutureTask.java:122)

6. at java.util.concurrent.FutureTask.get(FutureTask.java:192)

7. at com.developpez.lmauzaize.java.concurrence.thread.future.FutureRécupérationException.main(FutureRécupérationException.java:26)

8. Caused by: com.developpez.lmauzaize.java.concurrence.thread.future.FutureRécupérationException$1MonException

9. at com.developpez.lmauzaize.java.concurrence.thread.future.FutureRécupérationException$1.call(FutureRécupérationException.java:20)

10. at com.developpez.lmauzaize.java.concurrence.thread.future.FutureRécupérationException$1.call(FutureRécupérationException.java:1)

11. at java.util.concurrent.FutureTask.run(FutureTask.java:266)

12. at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)

13. at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)

14. at java.lang.Thread.run(Thread.java:745)

Sachez qu'il est également possible d'attendre le résultat pendant un temps défini. Une TimeoutException est levée si l'attente expire.

La méthode cancel permet d'annuler une tâche qui a été soumise.

- Si la tâche est dans la file d'attente, la tâche est simplement retirée de la file et n'est jamais exécutée.

Future - Annulation lors de l'attente (code)

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.future.FutureAnnulationAttente

2. Callable<Thread> tâche = new Callable<Thread>() {

3. public Thread call() throws Exception {

4. Logger.println("[Tâche] Début");

5. Logger.println("[Tâche] Fin");

6. return Thread.currentThread();

7. }

8. };

9. Callable<Thread> attente = new Callable<Thread>() {

10. public Thread call() throws Exception {

11. Logger.println("[Attente] Début");

12. Thread.sleep(3_000);

13. Logger.println("[Attente] Fin");

14. return Thread.currentThread();

15. }

16. };

17. ExecutorService executor = Executors.newSingleThreadExecutor();

19. // Monopolisation du service

20. executor.submit(attente);

22. // Soumission de la tâche à annuler

23. Future<Thread> future = executor.submit(tâche);

25. // Programme l'annulation

26. new Thread("Annulation") {

27. public void run() {

28. try {

29. Thread.sleep(2_000);

30. } catch (InterruptedException e) {

31. Logger.printStackTrace(e);

32. } finally {

33. Logger.println("Annulation de la tâche (%s)", future.cancel(false));

34. }

35. };

36. }.start();

38. // Récupération du résultat

39. try {

40. Thread thread = future.get();

41. Logger.println("La tâche a été exécutée sur %s", thread);

42. } catch (CancellationException e) {

43. Logger.println("La tâche a été annulée");

44. }

46. executor.shutdown();

Future - Annulation lors de l'attente (console)

1. 00:00:00.019 [pool-1-thread-1] [Attente] Début

2. 00:00:02.000 [main           ] La tâche a été annulée

3. 00:00:02.000 [Annulation     ] Annulation de la tâche (true)

4. 00:00:03.027 [pool-1-thread-1] [Attente] Fin

- Si la tâche est déjà en cours d'exécution et qu'il n'y a pas de demande d'interruption, la tâche continue son exécution, mais il n'est plus possible de récupérer son résultat.

Future - Annulation sans interruption (code)

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.future.FutureAnnulationSansInterruption

2. Callable<Thread> tâche = new Callable<Thread>() {

3. public Thread call() throws Exception {

4. Logger.println("[Attente] Début");

5. Thread.sleep(5_000);

6. Logger.println("[Attente] Fin");

7. return Thread.currentThread();

8. }

9. };

10. ExecutorService executor = Executors.newSingleThreadExecutor();

12. // Soumission de la tâche à annuler

13. Future<Thread> future = executor.submit(tâche);

15. // Programme l'annulation

16. Thread.sleep(2_000);

17. Logger.println("Annulation de la tâche (%s)", future.cancel(false));

19. // Récupération du résultat

20. try {

21. Thread thread = future.get();

22. Logger.println("La tâche a été exécutée sur %s", thread);

23. } catch (CancellationException e) {

24. Logger.println("La tâche a été annulée");

25. }

27. executor.shutdown();

Future - Annulation sans interruption (console)

1. 00:00:00.019 [pool-1-thread-1] [Attente] Début

2. 00:00:01.999 [main           ] Annulation de la tâche (true)

3. 00:00:02.000 [main           ] La tâche a été annulée

4. 00:00:05.027 [pool-1-thread-1] [Attente] Fin

- S'il y a en plus une demande d'interruption, le thread exécutant la tâche est interrompu. D'une part, la tâche est libre de traiter cette interruption comme il lui convient ; d'autre part, le thread continuera à exécuter les tâches qui suivent.

Future - Annulation avec interruption (code)

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.future.FutureAnnulationAvecInterruption

2. Callable<Thread> tâche = new Callable<Thread>() {

3. public Thread call() throws Exception {

4. Logger.println("[Attente] Début");

5. try {

6. Thread.sleep(5_000);

7. } catch (InterruptedException e) {

8. Logger.println("[Attente] La tâche a été annulée");

9. }

10. Logger.println("[Attente] Fin");

11. return Thread.currentThread();

12. }

13. };

14. Callable<Thread> suite = new Callable<Thread>() {

15. public Thread call() throws Exception {

16. Logger.println("[Suite] Début");

17. Logger.println("[Suite] Fin");

18. return Thread.currentThread();

19. }

20. };

21. ExecutorService executor = Executors.newSingleThreadExecutor();

22. // Soumission des tâches

23. Future<Thread> future1 = executor.submit(tâche);

24. Future<Thread> future2 = executor.submit(suite);

26. // Programme l'annulation

27. Thread.sleep(2_000);

28. Logger.println("Annulation de la tâche [Attente](%s)", future1.cancel(true));

30. // Récupération des résultats

31. try {

32. Thread thread = future1.get();

33. Logger.println("La tâche [Attente] a été exécutée sur %s", thread.getName());

34. } catch (CancellationException e) {

35. Logger.println("La tâche [Attente] a été annulée");

36. }

37. Thread thread = future2.get();

38. Logger.println("La tâche [Suite  ] a été exécutée sur %s", thread.getName());

40. executor.shutdown();

Future - Annulation avec interruption (console)

1. 00:00:00.020 [pool-1-thread-1] [Attente] Début

2. 00:00:02.000 [main           ] Annulation de la tâche [Attente](true)

3. 00:00:02.000 [pool-1-thread-1] [Attente] La tâche a été annulée

4. 00:00:02.001 [main           ] La tâche [Attente] a été annulée

5. 00:00:02.001 [pool-1-thread-1] [Attente] Fin

6. 00:00:02.002 [pool-1-thread-1] [Suite] Début

7. 00:00:02.003 [pool-1-thread-1] [Suite] Fin

8. 00:00:02.004 [main           ] La tâche [Suite  ] a été exécutée sur pool-1-thread-1

III-D - Fork/Join (1.7+)

Le framework Fork/Join est l'une des grandes nouveautés de Java 7. Il repose sur le principe « Diviser pour régner », c'est-à-dire découper le travail en petites tâches. Mais plutôt que découper le travail en tout petits morceaux que devront se disputer les threads, le principe est de diviser le traitement si la taille du problème est trop importante :

Fork/Join - principes

1. si (taille non critique) {

2. <exécuter le traitement>

3. } sinon {

4. <découper le traitement en deux parties>

5. <lancer l'exécution des deux parties>

6. }

Avec l'API Fork/Join, cela donne un gabarit de cette forme :

Fork/Join - patron de code

1. abstract class MonAction extends java.util.concurrent.RecursiveAction {

3. abstract boolean estTailleCritique();

4. abstract void    avant();

5. abstract void    traitement();

6. abstract void    après();

8. abstract java.util.concurrent.ForkJoinTask<Void> partie(int n);

10. @Override

11. protected void compute() {

12. avant();

14. if (!estTailleCritique()) {

15. traitement();

16. } else {

17. invokeAll(partie(1), partie(2));

18. }

20. après();

21. }

22. }

Concrètement, l'API se base sur la classe ForkJoinTask. Il est en revanche recommandé d'étendre RecursiveAction si votre traitement ne renvoie pas de résultat (void), sinon RecursiveTask.

Pour la démonstration, je vous propose un petit programme qui va simplement remplir un tableau d'entiers avec des nombres aléatoires :

Fork/Join - Démo (code)

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.forkjoin.ForkJoinDemo

2. class Randomizer {

3. int[] table = new int[1024 * 1024 * 256];

4. void random(int start, int end) {

5. ThreadLocalRandom random = ThreadLocalRandom.current();

6. for (int i = start; i <= end; i++) {

7. table[i] = random.nextInt();

8. }

9. }

10. void randomAll() {

11. random(0, table.length - 1);

12. table = null;

13. }

14. }

15. class RandomizerForkJoinAction extends RecursiveAction {

16. Randomizer randomizer;

17. int start, end;

18. RandomizerForkJoinAction() {

19. this.randomizer = new Randomizer();

20. this.start      = 0;

21. this.end        = randomizer.table.length - 1;

22. }

23. RandomizerForkJoinAction(Randomizer randomizer, int start, int end) {

24. this.randomizer = randomizer;

25. this.start      = start;

26. this.end        = end;

27. }

29. @Override

30. protected void compute() {

31. if ((end - start) <= 1024 * 1024) {

32. randomizer.random(start, end);

33. } else {

34. int mid = (this.end + this.start) /2;

35. invokeAll(

36. new RandomizerForkJoinAction(randomizer, start, mid),

37. new RandomizerForkJoinAction(randomizer, mid+1, end)

38. );

39. }

40. }

41. }

43. long start;

45. Logger.println("Monothread - début");

46. start = System.currentTimeMillis();

47. new Randomizer().randomAll();

48. Logger.println("Monothread - fin (%d ms)", System.currentTimeMillis() - start);

50. Logger.println("Fork/Join - début");

51. start = System.currentTimeMillis();

52. new ForkJoinPool().invoke(new RandomizerForkJoinAction());

53. Logger.println("Fork/Join - fin (%d ms)", System.currentTimeMillis() - start);

Fork/Join - Démo (console)

1. 00:00:00.020 [main           ] Monothread - début

2. 00:00:00.682 [main           ] Monothread - fin (652 ms)

3. 00:00:00.682 [main           ] Fork/Join - début

4. 00:00:01.022 [main           ] Fork/Join - fin (340 ms)

Mais pourquoi invoquer les sous-tâches alors que le thread courant va être en attente de leur résultat ? Et bien, c'est ce que va faire la méthode invokeAll ! Elle va mettre en file les tâches sauf une et l'exécuter, puis attendre que les sous-tâches se terminent :

Fork/Join - Tâche

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.forkjoin.ForkJoinTâche

2. abstract class ForkJoinTâche extends RecursiveAction {

3. String  nom;

4. ForkJoinTâche() {

5. this.nom = getClass().getSimpleName();

6. }

7. ForkJoinTâche(int n) {

8. this.nom = getClass().getSimpleName() + " " + n;

9. }

10. ForkJoinTâche(String nom) {

11. this.nom = nom;

12. }

13. protected void compute() {

14. Logger.println("[%s] Début", nom);

15. calcul();

16. Logger.println("[%s] Fin", nom);

17. }

18. abstract void calcul();

19. }

Fork/Join - Première tâche sur le même thread (code)

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.forkjoin.ForkJoinMemeThread

2. class SousTâche extends ForkJoinTâche {

3. void calcul() {

4. TimeUnit.SECONDS.sleep(2);

5. }

6. }

7. class SousTâche1 extends SousTâche {}

8. class SousTâche2 extends SousTâche {}

9. class Tâche extends ForkJoinTâche {

10. void calcul() {

11. invokeAll(new SousTâche1(), new SousTâche2());

12. }

13. }

15. new ForkJoinPool().invoke(new Tâche());

Fork/Join - Première tâche sur le même thread (console)

1. 00:00:00.020 [ForkJoinPool-1-worker-1] [Tâche] Début

2. 00:00:00.028 [ForkJoinPool-1-worker-1] [SousTâche1] Début

3. 00:00:00.028 [ForkJoinPool-1-worker-2] [SousTâche2] Début

4. 00:00:02.028 [ForkJoinPool-1-worker-1] [SousTâche1] Fin

5. 00:00:02.028 [ForkJoinPool-1-worker-2] [SousTâche2] Fin

6. 00:00:02.029 [ForkJoinPool-1-worker-1] [Tâche] Fin

On remarque que Tâche et Sous-tâche 1 ont été exécutées par ForkJoinPool-1-worker-1 tandis que Sous-tâche 2 a été exécutée par ForkJoinPool-1-worker-2.

Mais que se passe-t-il s'il n'y a pas assez de threads pour exécuter les sous-tâches ? Ne risque-t-on pas d'avoir un deadlock ?

Fork/Join - Pas assez de thread (code)

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.forkjoin.ForkJoinManqueThread

2. new ForkJoinPool(1).invoke(new Tâche());

Fork/Join - Pas assez de thread (console)

1. [ForkJoinPool-1-worker-1] Début de Tâche

2. [ForkJoinPool-1-worker-1] Début de Sous-tâche 1

3. [ForkJoinPool-1-worker-1] Fin   de Sous-tâche 1

4. [ForkJoinPool-1-worker-1] Début de Sous-tâche 2

5. [ForkJoinPool-1-worker-1] Fin   de Sous-tâche 2

6. [ForkJoinPool-1-worker-1] Fin   de Tâche

L'exemple est similaire au précédent, excepté que l'on a défini un niveau de parallélisme à 1. On remarque alors que les deux sous-tâches ont été exécutées par le thread qui était en train d'exécuter la tâche principale.

En revanche, il est possible de saturer votre pool si vous effectuez des tâches bloquantes :

Fork/Join - Tâches bloquées (code)

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.forkjoin.ForkJoinBloqué

2. class Verrou {

3. void prendre(); // Attend jusqu'à obtention verrou

4. void libérer(); // Libère le verrou

5. }

6. Verrou verrou = new Verrou();

8. class Bloqueur extends ForkJoinTâche {

9. void calcul() {

10. Logger.println("[%s] Pose du verrou", nom);

11. verrou.prendre();

12. Logger.println("[%s] Calcul", nom);

13. TimeUnit.SECONDS.sleep(5);

14. verrou.libérer();

15. }

16. }

17. class SousTâche extends ForkJoinTâche {

18. SousTâche(int n) {

19. super(n);

20. }

21. void calcul() {

22. TimeUnit.SECONDS.sleep(1);

23. }

24. }

26. class Tâche extends ForkJoinTâche {

27. protected void calcul() {

28. List<ForkJoinTask> actions = new ArrayList<>();

29. actions.add(new Bloqueur());

30. for (int i = 0; i < 5; i++) {

31. actions.add(new SousTâche(i));

32. }

33. invokeAll(actions);

34. }

35. }

37. Logger.println("Pose du verrou");

38. verrou.prendre();

39. Logger.println("Soumission");

40. Tâche tâche = new Tâche();

41. new ForkJoinPool(1).execute(tâche);

42. Logger.println("Pause");

43. TimeUnit.SECONDS.sleep(5);

44. Logger.println("Libération du verrou");

45. verrou.libérer();

46. tâche.join();

Fork/Join - Tâches bloquées (console)

1. 00:00:00.018 [main           ] Pose du verrou

2. 00:00:00.026 [main           ] Soumission

3. 00:00:00.029 [main           ] Pause

4. 00:00:00.029 [ForkJoinPool-1-worker-1] [Tâche] Début

5. 00:00:00.030 [ForkJoinPool-1-worker-1] [Bloqueur] Début

6. 00:00:00.030 [ForkJoinPool-1-worker-1] [Bloqueur] Pose du verrou

7. 00:00:05.029 [main           ] Libération du verrou

8. 00:00:05.029 [ForkJoinPool-1-worker-1] [Bloqueur] Calcul

9. 00:00:10.030 [ForkJoinPool-1-worker-1] [Bloqueur] Fin

10. 00:00:10.030 [ForkJoinPool-1-worker-1] [SousTâche 0] Début

11. 00:00:11.031 [ForkJoinPool-1-worker-1] [SousTâche 0] Fin

12. 00:00:11.031 [ForkJoinPool-1-worker-1] [SousTâche 1] Début

13. 00:00:12.032 [ForkJoinPool-1-worker-1] [SousTâche 1] Fin

14. 00:00:12.032 [ForkJoinPool-1-worker-1] [SousTâche 2] Début

15. 00:00:13.033 [ForkJoinPool-1-worker-1] [SousTâche 2] Fin

16. 00:00:13.033 [ForkJoinPool-1-worker-1] [SousTâche 3] Début

17. 00:00:14.034 [ForkJoinPool-1-worker-1] [SousTâche 3] Fin

18. 00:00:14.034 [ForkJoinPool-1-worker-1] [SousTâche 4] Début

19. 00:00:15.036 [ForkJoinPool-1-worker-1] [SousTâche 4] Fin

20. 00:00:15.036 [ForkJoinPool-1-worker-1] [Tâche] Fin

Il est dommage qu'il ait fallu attendre que la tâche bloquante ait accès à la ressource pour exécuter les tâches qui n'en dépendaient pas.

Pour résoudre ce problème, Fork/Join framework propose l'interface ManagedBlocker. L'interface se compose de deux méthodes :

- isReleasable : elle doit renvoyer vrai (true) lorsqu'il n'est pas/plus nécessaire de bloquer. Cela consiste généralement à essayer d'obtenir l'accès à la ressource de manière non bloquante ;

- block : gère l'appel bloquant et renvoie vrai (true) si l'exécution peut se poursuivre.

Une fois l'instance créée, elle doit être passée en paramètre de la méthode statique ForkJoinPool.managedBlock. Retravaillons notre exemple :

Fork/Join - ManagedBlocker (code)

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.forkjoin.ForkJoinManagedBlocker

2. class Verrou {

3. // ...

4. boolean tenter(); // Essaye de prendre le verrou et renvoie le résultat.

5. }

7. Verrou verrou = new Verrou();

8. class Verrouilleur implements ManagedBlocker, AutoCloseable {

9. boolean verrouillé = false;

10. Verrouilleur() {

11. ForkJoinPool.managedBlock(this);

12. }

13. public boolean isReleasable() {

14. if (!verrouillé) {

15. verrouillé = verrou.tenter();

16. }

17. return verrouillé;

18. }

19. public boolean block() {

20. if (!verrouillé) {

21. verrou.prendre();

22. verrouillé = true;

23. }

24. return verrouillé;

25. }

26. public void close() {

27. verrou.libérer();

28. verrouillé = false;

29. }

30. }

32. class Bloqueur extends ForkJoinTâche {

33. void calcul() {

34. Logger.println("[%s] Pose du verrou", nom);

35. try (Verrouilleur verrouilleur = new Verrouilleur()) {

36. Logger.println("[%s] Calcul", nom);

37. TimeUnit.SECONDS.sleep(2);

38. }

39. }

40. }

41. class SousTâche extends ForkJoinTâche {

42. SousTâche(int n) {

43. super(n);

44. }

45. void calcul() {

46. TimeUnit.SECONDS.sleep(1);

47. }

48. }

49. class Tâche extends ForkJoinTâche {

50. protected void calcul() {

51. List<ForkJoinTask> actions = new ArrayList<>();

52. actions.add(new Bloqueur());

53. for (int i = 0; i < 10; i++) {

54. actions.add(new SousTâche(i));

55. }

56. invokeAll(actions);

57. }

58. }

60. Logger.println("Pose du verrou");

61. verrou.prendre();

62. Logger.println("Soumission");

63. Tâche tâche = new Tâche();

64. new ForkJoinPool(1).execute(tâche);

65. Logger.println("Pause");

66. TimeUnit.SECONDS.sleep(3);

67. Logger.println("Libération du verrou");

68. verrou.libérer();

69. tâche.join();

Fork/Join - ManagedBlocker (console - Java 8)

1. 00:00:00.027 [main           ] Pose du verrou

2. 00:00:00.037 [main           ] Soumission

3. 00:00:00.040 [main           ] Pause

4. 00:00:00.041 [ForkJoinPool-1-worker-1] [Tâche] Début

5. 00:00:00.043 [ForkJoinPool-1-worker-1] [Bloqueur] Début

6. 00:00:00.043 [ForkJoinPool-1-worker-1] [Bloqueur] Pose du verrou

7. 00:00:00.047 [ForkJoinPool-1-worker-0] [SousTâche 9] Début

8. 00:00:01.047 [ForkJoinPool-1-worker-0] [SousTâche 9] Fin

9. 00:00:01.047 [ForkJoinPool-1-worker-0] [SousTâche 8] Début

10. 00:00:02.048 [ForkJoinPool-1-worker-0] [SousTâche 8] Fin

11. 00:00:02.048 [ForkJoinPool-1-worker-0] [SousTâche 7] Début

12. 00:00:03.041 [main           ] Libération du verrou

13. 00:00:03.041 [ForkJoinPool-1-worker-1] [Bloqueur] Calcul

14. 00:00:03.049 [ForkJoinPool-1-worker-0] [SousTâche 7] Fin

15. 00:00:03.049 [ForkJoinPool-1-worker-0] [SousTâche 6] Début

16. 00:00:04.050 [ForkJoinPool-1-worker-0] [SousTâche 6] Fin

17. 00:00:04.050 [ForkJoinPool-1-worker-0] [SousTâche 5] Début

18. 00:00:05.043 [ForkJoinPool-1-worker-1] [Bloqueur] Fin

19. 00:00:05.043 [ForkJoinPool-1-worker-1] [SousTâche 0] Début

20. 00:00:05.052 [ForkJoinPool-1-worker-0] [SousTâche 5] Fin

21. 00:00:05.052 [ForkJoinPool-1-worker-0] [SousTâche 4] Début

22. 00:00:06.044 [ForkJoinPool-1-worker-1] [SousTâche 0] Fin

23. 00:00:06.044 [ForkJoinPool-1-worker-1] [SousTâche 1] Début

24. 00:00:06.053 [ForkJoinPool-1-worker-0] [SousTâche 4] Fin

25. 00:00:06.053 [ForkJoinPool-1-worker-0] [SousTâche 3] Début

26. 00:00:07.045 [ForkJoinPool-1-worker-1] [SousTâche 1] Fin

27. 00:00:07.045 [ForkJoinPool-1-worker-1] [SousTâche 2] Début

28. 00:00:07.053 [ForkJoinPool-1-worker-0] [SousTâche 3] Fin

29. 00:00:08.046 [ForkJoinPool-1-worker-1] [SousTâche 2] Fin

30. 00:00:08.046 [ForkJoinPool-1-worker-1] [Tâche] Fin

On remarque alors deux choses. La première, c'est que le pool a alloué un nouveau thread pour traiter les tâches.

La deuxième chose, c'est que les sous-tâches n'ont pas été exécutées selon l'ordre de soumission à cause du (grâce au ?) work-stealing algorithm. Le deuxième thread a pris les tâches selon leur ordre inverse pour limiter l'utilisation de verrou pour accéder à la liste de tâches. En utilisant les deux extrémités de la liste des tâches, on limite les contentions.

Il peut-être intéressant de savoir que Phaser (1.7+) et l'API CompletableFuture (1.8+) exploite ce mécanisme pour limiter les contentions.

Il est important de noter que le comportement avec une JVM 1.7 n'est pas exactement le même. En effet, dans la version précédente, le nouveau thread ne traitait les tâches que le temps du blocage. Comme le montrent les traces ci-dessous.

Fork/Join - ManagedBlocker (console - Java 7)

1. 00:00:00.011 [main           ] Pose du verrou

2. 00:00:00.021 [main           ] Soumission

3. 00:00:00.023 [main           ] Pause

4. 00:00:00.023 [ForkJoinPool-1-worker-1] [Tâche] Début

5. 00:00:00.024 [ForkJoinPool-1-worker-1] [Bloqueur] Début

6. 00:00:00.025 [ForkJoinPool-1-worker-1] [Bloqueur] Pose du verrou

7. 00:00:00.025 [ForkJoinPool-1-worker-2] [SousTâche 9] Début

8. 00:00:01.026 [ForkJoinPool-1-worker-2] [SousTâche 9] Fin

9. 00:00:01.026 [ForkJoinPool-1-worker-2] [SousTâche 8] Début

10. 00:00:02.027 [ForkJoinPool-1-worker-2] [SousTâche 8] Fin

11. 00:00:02.027 [ForkJoinPool-1-worker-2] [SousTâche 7] Début

12. 00:00:03.024 [main           ] Libération du verrou

13. 00:00:03.024 [ForkJoinPool-1-worker-1] [Bloqueur] Calcul

14. 00:00:03.028 [ForkJoinPool-1-worker-2] [SousTâche 7] Fin

15. 00:00:05.025 [ForkJoinPool-1-worker-1] [Bloqueur] Fin

16. 00:00:05.025 [ForkJoinPool-1-worker-1] [SousTâche 0] Début

17. 00:00:06.026 [ForkJoinPool-1-worker-1] [SousTâche 0] Fin

18. 00:00:06.026 [ForkJoinPool-1-worker-1] [SousTâche 1] Début

19. 00:00:07.027 [ForkJoinPool-1-worker-1] [SousTâche 1] Fin

20. 00:00:07.027 [ForkJoinPool-1-worker-1] [SousTâche 2] Début

21. 00:00:08.028 [ForkJoinPool-1-worker-1] [SousTâche 2] Fin

22. 00:00:08.028 [ForkJoinPool-1-worker-1] [SousTâche 3] Début

23. 00:00:09.029 [ForkJoinPool-1-worker-1] [SousTâche 3] Fin

24. 00:00:09.029 [ForkJoinPool-1-worker-1] [SousTâche 4] Début

25. 00:00:10.030 [ForkJoinPool-1-worker-1] [SousTâche 4] Fin

26. 00:00:10.030 [ForkJoinPool-1-worker-1] [SousTâche 5] Début

27. 00:00:11.031 [ForkJoinPool-1-worker-1] [SousTâche 5] Fin

28. 00:00:11.031 [ForkJoinPool-1-worker-1] [SousTâche 6] Début

29. 00:00:12.032 [ForkJoinPool-1-worker-1] [SousTâche 6] Fin

30. 00:00:12.032 [ForkJoinPool-1-worker-1] [Tâche] Fin

Cependant, il est important de savoir que l'utilisation de ManagedBlocker est surtout dédiée à limiter les deadlocks. En effet, chaque worker bloqué n'est pas remplacé. Des workers supplémentaires sont uniquement alloués si le pool le permet (limite du nombre de threads) et si TOUS les workers sont bloqués :

Fork/Join - Limite des ManagedBlockers (code)

1. //com.developpez.lmauzaize.java.concurrence.ch01_gestion_threads.forkjoin.ForkJoinManagedBlockerLimite

2. class TâcheBloquante extends ForkJoinTâche {

3. Verrou verrou1 = new Verrou();

4. Verrou verrou2 = new Verrou();

5. TâcheBloquante(int n) {

6. super(n);

7. }

8. protected void calcul() {

9. Logger.println("[%s] Attente non-managed", nom);

10. verrou1.prendre();

12. Logger.println("[%s] Passage en mode 'managed'", nom);

13. try (Verrouilleur verrouilleur = new Verrouilleur(verrou2)) {}

14. }

15. }

16. class SousTâche extends ForkJoinTâche {

17. SousTâche(int n) {

18. super(n);

19. }

20. @Override

21. void calcul() {

22. TimeUnit.SECONDS.sleep(1);

23. }

24. }

25. class Tâche extends ForkJoinTâche {

26. @Override

27. void calcul() {

28. List<ForkJoinTâche> soustâches = new ArrayList<>();

29. for (int i = 1; i <= 5; i++) {

30. soustâches.add(new SousTâche(i));

31. }

32. invokeAll(soustâches);

33. }

34. }

36. int parallélisme = 2;

37. ForkJoinPool pool = new ForkJoinPool(parallélisme);

39. Logger.println("Saturation du pool");

40. List<TâcheBloquante> bloqués = new ArrayList<>(parallélisme);

41. for (int i = 1; i <= parallélisme; i++) {

42. TâcheBloquante bloqué = new TâcheBloquante(i);

43. bloqué.verrou1.prendre();

44. bloqué.verrou2.prendre();

45. pool.submit(bloqué);

46. bloqués.add(bloqué);

47. }

49. Logger.println("Ajout de tâches supplémentaires");

50. Tâche tâche = new Tâche();

51. pool.submit(tâche);

53. for (TâcheBloquante bloqué : bloqués) {

54. TimeUnit.SECONDS.sleep(5);

55. bloqué.verrou1.libérer();

56. }

58. tâche.join();

59. Logger.println("Libération des tâches bloquantes");

60. for (TâcheBloquante bloqué : bloqués) {

61. bloqué.verrou2.libérer();

62. bloqué.join();

63. }

Fork/Join - Limite des ManagedBlockers (console)

1. 00:00:00.020 [main           ] Saturation du pool

2. 00:00:00.031 [main           ] Ajout de tâches supplémentaires

3. 00:00:00.031 [ForkJoinPool-1-worker-0] [TâcheBloquante 2] Début

4. 00:00:00.031 [ForkJoinPool-1-worker-0] [TâcheBloquante 2] Attente non-managed

5. 00:00:00.031 [ForkJoinPool-1-worker-1] [TâcheBloquante 1] Début

6. 00:00:00.032 [ForkJoinPool-1-worker-1] [TâcheBloquante 1] Attente non-managed

7. 00:00:05.032 [ForkJoinPool-1-worker-1] [TâcheBloquante 1] Passage en mode 'managed'

8. 00:00:10.032 [ForkJoinPool-1-worker-0] [TâcheBloquante 2] Passage en mode 'managed'

9. 00:00:10.033 [ForkJoinPool-1-worker-2] [Tâche] Début

10. 00:00:10.034 [ForkJoinPool-1-worker-2] [SousTâche 1] Début

11. 00:00:11.035 [ForkJoinPool-1-worker-2] [SousTâche 1] Fin

12. 00:00:11.035 [ForkJoinPool-1-worker-2] [SousTâche 2] Début

13. 00:00:12.036 [ForkJoinPool-1-worker-2] [SousTâche 2] Fin

14. 00:00:12.036 [ForkJoinPool-1-worker-2] [SousTâche 3] Début

15. 00:00:13.037 [ForkJoinPool-1-worker-2] [SousTâche 3] Fin

16. 00:00:13.037 [ForkJoinPool-1-worker-2] [SousTâche 4] Début

17. 00:00:14.038 [ForkJoinPool-1-worker-2] [SousTâche 4] Fin

18. 00:00:14.038 [ForkJoinPool-1-worker-2] [SousTâche 5] Début

19. 00:00:15.039 [ForkJoinPool-1-worker-2] [SousTâche 5] Fin

20. 00:00:15.039 [ForkJoinPool-1-worker-2] [Tâche] Fin

21. 00:00:15.040 [main           ] Libération des tâches bloquantes

22. 00:00:15.040 [ForkJoinPool-1-worker-1] [TâcheBloquante 1] Fin

23. 00:00:15.041 [ForkJoinPool-1-worker-0] [TâcheBloquante 2] Fin

On a commencé par saturer le pool avec des tâches bloquantes, puis on a ajouté une tâche qui produisait cinq sous-tâches. On constate alors que rien ne se produit.
Ensuite, on passe peu à peu les tâches bloquantes en mode « managed ». Tant qu'elles ne sont pas toutes passées en mode « managed », il ne se passe toujours rien.
Dès que la dernière bloquante est passée en mode « managed », le pool alloue uniquement un troisième worker qui traite la tâche (et toutes ses filles).
Finalement on libère les tâches bloquantes.

Si vous devez utiliser des opérations bloquantes (verrous ou non), il est recommandé de s'orienter vers des API asynchrones.

IV - Conclusion

Au cours de ce premier article, nous avons vu comment sont représentés les threads en Java et comment interagir avec eux. Ceci pose les fondations de la programmation parallèle, mais les prochains articles seront entièrement dédiés à gérer la concurrence dans votre code à l'aide de l'API native (intégrée au langage) et des outils fournis par Java SE.

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

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.