Framework Fork/Join de Java | W3Docs Learn Java

Un pool de threads ordinaire est excellent pour « de nombreuses tâches indépendantes ». Il n'est pas adapté à « une grande tâche qui peut être divisée récursivement en versions plus petites d'elle-même ». Pour cette deuxième forme — le travail en diviser-pour-régner — Java dispose d'un exécuteur spécialisé : le ForkJoinPool. C'est le pool derrière parallelStream, CompletableFuture.supplyAsync (lorsqu'aucun exécuteur n'est fourni), et tout code que vous écrivez avec RecursiveTask/RecursiveAction.

La technique qui rend ForkJoinPool spécial est le vol de tâches : chaque worker possède sa propre deque, et lorsque sa propre deque est vide, il vole une tâche depuis le bas de la deque d'un autre worker. Le résultat est un équilibrage de charge automatique — les workers rapides aident les workers lents sans aucune coordination.

Quand l'utiliser

Fork/join est l'outil adapté pour :

La récursion en diviser-pour-régner. Tri rapide, tri fusion, parcours d'arbres, algorithmes numériques récursifs, multiplication de matrices par halvage.
Le travail lié au CPU qui est parallélisable en parties approximativement égales.
Un travail dont la granularité s'adapte : diviser si le bloc est grand, exécuter directement s'il est petit.

C'est le mauvais outil pour :

Le travail lié aux I/O. Un worker bloqué ne vole pas — et la taille par défaut du pool est le nombre de vos CPU. Bloquer un worker revient à perdre un cœur.
Des tâches indépendantes sans relation. Un ThreadPoolExecutor ordinaire est plus simple et tout aussi rapide pour cette forme.
Des tâches qui dépendent d'un planning externe fixe. Utilisez ScheduledExecutorService.

Un modèle mental pratique : si vous utiliseriez parallelStream pour cela, fork/join est la même forme exprimée directement. (Fork/join est arrivé en Java 7 ; parallelStream en Java 8 a été construit par-dessus.)

Les trois classes

ForkJoinPool pool;                                    // the executor
RecursiveTask<V>;                                     // an abstract task returning V
RecursiveAction;                                      // an abstract task returning nothing

Vous étendez RecursiveTask ou RecursiveAction, redéfinissez compute(), décidez dans compute() s'il faut diviser ou exécuter le travail directement, et appelez fork()/join() sur les sous-tâches.

class Sum extends RecursiveTask<Long> {
  private static final int THRESHOLD = 1000;
  private final long[] data;
  private final int lo, hi;

  Sum(long[] data, int lo, int hi) {
    this.data = data; this.lo = lo; this.hi = hi;
  }

  @Override
  protected Long compute() {
    int len = hi - lo;
    if (len <= THRESHOLD) {
      long s = 0;
      for (int i = lo; i < hi; i++) s += data[i];
      return s;
    }
    int mid = lo + len / 2;
    Sum left  = new Sum(data, lo, mid);
    Sum right = new Sum(data, mid, hi);
    left.fork();                                      // schedule left to run on another worker
    long rightResult = right.compute();               // run right on this worker (avoid extra task)
    long leftResult  = left.join();                   // wait for left
    return leftResult + rightResult;
  }
}

ForkJoinPool pool = new ForkJoinPool();
long total = pool.invoke(new Sum(data, 0, data.length));

La forme — vérifier le seuil → diviser → forker une moitié → calculer l'autre → joindre — est l'idiome fork/join canonique. L'astuce « calculer une moitié ici plutôt que de forker les deux » évite de créer une tâche inutile et constitue un gain petit mais réel.

Le seuil est important

La décision la plus importante : quand arrêter de diviser. Un seuil trop petit crée des milliers de tâches pour des morceaux triviaux — la surcharge domine le travail. Un seuil trop grand empêche d'exploiter pleinement les cœurs — de nombreux workers restent inactifs pendant qu'un seul traite un gros bloc.

Règles empiriques :

Le corps de la tâche devrait prendre au moins 10 microsecondes. En dessous, la surcharge de gestion des tâches est comparable au travail lui-même.
Faites du seuil une constante ajustable. 100, 1000, 10000 sont typiques pour les tableaux de primitives ; le bon nombre dépend du coût par élément.
Pour de très petites entrées, revenez à une implémentation purement séquentielle. La surcharge de division et de fork est gaspillée sur des entrées qui tiennent dans le cache.

`fork()`, `join()`, `invoke()`

Les trois opérations sur un RecursiveTask :

Méthode	Comportement
`fork()`	Planifie la tâche dans le pool courant ; retourne immédiatement
`join()`	Attend la tâche et retourne son résultat (ou relance ce qu'elle a lancé)
`invoke()`	Combinaison de `fork` + `join` pour ce thread — synchrone
`compute()`	Exécute le corps directement sur le thread appelant (sans fork)

Dans le schéma ci-dessus, left.fork(); right.compute(); left.join(); fait ce qu'il faut — forker une moitié vers un autre worker, exécuter l'autre moitié ici, puis attendre la fin du fork.

Vous ne devriez pas écrire left.fork(); right.fork(); left.join(); right.join();. Le côté droit est forké et le worker courant attend — il n'y a aucun fil d'exécution disponible pour réellement exécuter right jusqu'à ce que le worker atteigne join. La combinaison gaspille le créneau temporel du worker courant.

Le pool commun

ForkJoinPool.commonPool() est un pool partagé à l'échelle de la JVM, dimensionné à Runtime.getRuntime().availableProcessors() - 1 par défaut. Il alimente :

Stream.parallelStream()
CompletableFuture.supplyAsync(supplier) (la surcharge sans exécuteur)
Arrays.parallelSort()

Vous pouvez configurer la taille du pool commun via la propriété système java.util.concurrent.ForkJoinPool.common.parallelism au démarrage de la JVM. Vous ne devriez pas utiliser le pool commun pour les I/O — un seul appel bloquant occupe un worker que toute la JVM partage.

Le vol de tâches en images

worker-1 deque:  [t1 t2 t3 t4]            (it forked these; t4 just got pushed)
worker-2 deque:  []                       (empty — workers steal)
worker-3 deque:  [t10 t11]                (still has its own)

worker-2 finds its deque empty; steals t1 from the BOTTOM of worker-1's deque
worker-1 keeps pulling its own tasks from the TOP

La file double extrémité est au cœur de la conception : les workers poussent et retirent d'un côté (LIFO — localité de référence pour les hits de cache), les voleurs prennent de l'autre côté (FIFO — contention minimale avec le propriétaire). C'est pourquoi fork/join fonctionne si bien : les workers se contentent rarement sur les structures de données des autres, même sous forte charge.

Un exemple travaillé : somme parallèle vs. séquentielle

Le programme ci-dessous somme un tableau de 10 millions d'éléments de deux façons — boucle séquentielle et récursion fork/join — et affiche le temps réel pour chacune.

java— editable, runs on the server

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public class ForkJoinDemo {
  public static void main(String[] args) {
    int n = 10_000_000;
    long[] data = new long[n];
    for (int i = 0; i < n; i++) data[i] = (i * 2654435761L) & 0xFFFF;     // some non-trivial values

int cores = Runtime.getRuntime().availableProcessors();
    System.out.println("cores: " + cores);
    System.out.println("common pool parallelism: " + ForkJoinPool.commonPool().getParallelism());

// 1. Serial baseline
    long t1 = System.nanoTime();
    long serialSum = 0;
    for (int i = 0; i < n; i++) serialSum += data[i];
    long serialMs = (System.nanoTime() - t1) / 1_000_000;
    System.out.printf("%nserial:    sum=%d   %d ms%n", serialSum, serialMs);

// 2. Fork/Join
    long t2 = System.nanoTime();
    long pjSum = ForkJoinPool.commonPool().invoke(new Sum(data, 0, n));
    long pjMs = (System.nanoTime() - t2) / 1_000_000;
    System.out.printf("fork/join: sum=%d   %d ms   (about %.1fx faster)%n",
        pjSum, pjMs, Math.max(1.0, (double) serialMs / Math.max(1, pjMs)));

System.out.println("\nsums equal? " + (serialSum == pjSum));

// 3. Threshold matters: too-small THRESHOLD makes it slower
    long t3 = System.nanoTime();
    long tinySum = ForkJoinPool.commonPool().invoke(new SumTiny(data, 0, n));
    long tinyMs = (System.nanoTime() - t3) / 1_000_000;
    System.out.printf("fork/join (threshold=10): sum=%d   %d ms  -- excessive task overhead%n",
        tinySum, tinyMs);
  }

// Sensible threshold
  static class Sum extends RecursiveTask<Long> {
    static final int THRESHOLD = 100_000;
    private final long[] data;
    private final int lo, hi;
    Sum(long[] data, int lo, int hi) { this.data = data; this.lo = lo; this.hi = hi; }
    @Override protected Long compute() {
      int len = hi - lo;
      if (len <= THRESHOLD) {
        long s = 0;
        for (int i = lo; i < hi; i++) s += data[i];
        return s;
      }
      int mid = lo + len / 2;
      Sum left  = new Sum(data, lo, mid);
      Sum right = new Sum(data, mid, hi);
      left.fork();
      long r = right.compute();
      long l = left.join();
      return l + r;
    }
  }

// Threshold WAY too small — millions of micro-tasks; overhead dominates
  static class SumTiny extends RecursiveTask<Long> {
    static final int THRESHOLD = 10;
    private final long[] data;
    private final int lo, hi;
    SumTiny(long[] data, int lo, int hi) { this.data = data; this.lo = lo; this.hi = hi; }
    @Override protected Long compute() {
      int len = hi - lo;
      if (len <= THRESHOLD) {
        long s = 0;
        for (int i = lo; i < hi; i++) s += data[i];
        return s;
      }
      int mid = lo + len / 2;
      SumTiny left  = new SumTiny(data, lo, mid);
      SumTiny right = new SumTiny(data, mid, hi);
      left.fork();
      long r = right.compute();
      long l = left.join();
      return l + r;
    }
  }
}

Ce qu'il faut retenir de l'exécution :

La version fork/join était plusieurs fois plus rapide que la boucle séquentielle. Sur une machine à N cœurs, la limite supérieure est environ N× — le nombre réel était inférieur car la JVM, le GC et les autres threads JVM voulaient aussi du CPU, et le travail du seuil n'est pas parfaitement équilibré. Malgré tout, une accélération substantielle pour quelques lignes de code récursif.
Les deux sommes étaient égales. C'est le test de correction par partition et fusion : chaque feuille a sommé sa tranche non chevauchante ; l'étape de combinaison (l + r) les a additionnées ; pas de double comptage ni de courses aux données car chaque feuille écrivait dans sa propre variable locale.
La variante SumTiny avec le seuil 10 était plus lente que la boucle séquentielle. Avec 10 millions d'éléments divisés en blocs de 10, vous créez environ 2 millions de tâches — et la surcharge de gestion des tâches dépasse le travail d'addition réel. Le seuil est un vrai paramètre ; mesurez-le sur des entrées représentatives.
Le schéma left.fork(); long r = right.compute(); long l = left.join(); utilisait une tâche de moins que fork(); fork(); join(); join();. Le worker courant a du temps libre pendant le compute() — l'utiliser pour une des moitiés économise une allocation de tâche entière. C'est le gain petit mais cumulatif sur de nombreuses charges de travail réelles.
ForkJoinPool.commonPool() était l'exécuteur utilisé dans cette démo. Pour une exécution ponctuelle, le pool commun convient. Pour un programme longue durée qui mélange le travail fork/join avec des appels parallel sur des streams et des futures asynchrones, donnez la charge lourde fork/join son propre pool — le pool commun est conçu pour de courtes rafales, pas pour un calcul intensif en régime permanent.

Prochaine étape

Le prochain chapitre, Collections concurrentes Java, couvre les structures de données conçues pour être sollicitées par plusieurs threads — ConcurrentHashMap, CopyOnWriteArrayList, BlockingQueue, et le reste de java.util.concurrent.

Pratique

Dans `compute()` d'un `RecursiveTask`, vous avez deux sous-tâches `left` et `right`. Quel schéma d'appel est l'idiome fork/join canonique ?

`left.fork(); long r = right.compute(); long l = left.join();` — forker une moitié, calculer l'autre sur le worker courant, joindre le fork`left.fork(); right.fork(); long l = left.join(); long r = right.join();` — forker les deux et joindre les deux`left.invoke(); right.invoke();` — les invocations s'exécutent de façon synchrone en parallèle automatiquement`left.start(); right.start(); left.join(); right.join();` — identique à l'API `Thread` ordinaire