Framework Fork/Join de Java | W3Docs Learn Java

Un pool de hilos convencional es ideal para "muchas tareas independientes". No es la mejor opción para "una tarea grande que puede dividirse recursivamente en versiones más pequeñas de sí misma". Para esa segunda forma — trabajo de divide y vencerás — Java dispone de un executor especializado: el ForkJoinPool. Es el pool que hay detrás de parallelStream, CompletableFuture.supplyAsync (cuando no se proporciona un executor) y cualquier código que escribas con RecursiveTask/RecursiveAction.

El truco que hace especial al ForkJoinPool es el work-stealing: cada worker tiene su propia cola doble, y cuando esta se vacía roba una tarea desde el extremo inferior de la cola de otro worker. El resultado es un equilibrio de carga automático: los workers rápidos ayudan a los lentos sin necesidad de coordinación.

Cuándo utilizarlo

Fork/join es la herramienta adecuada para:

Divide y vencerás recursivo. Quicksort, mergesort, recorrido de árboles, algoritmos numéricos recursivos, multiplicación de matrices por división a la mitad.
Trabajo intensivo en CPU que puede paralelizarse en partes aproximadamente iguales.
Trabajo cuya granularidad se adapta: divide si el fragmento es grande, procesa directamente si es pequeño.

No es la herramienta adecuada para:

Trabajo limitado por I/O. Un worker bloqueado no roba — y el tamaño predeterminado del pool es el número de CPUs. Bloquear un worker equivale a perder un núcleo.
Tareas independientes no relacionadas. Un ThreadPoolExecutor convencional es más sencillo e igual de rápido para ese patrón.
Tareas que dependen de un calendario externo fijo. Usa ScheduledExecutorService.

Un modelo mental útil: si recurrirías a un parallelStream para esto, fork/join es la misma forma expresada directamente. (Fork/join llegó en Java 7; parallelStream en Java 8 se construyó sobre él.)

Las tres clases

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

Extiendes RecursiveTask o RecursiveAction, sobreescribes compute(), decides dentro de compute() si dividir o realizar el trabajo directamente, y llamas a fork()/join() sobre las subtareas.

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 forma — comprobar umbral → dividir → hacer fork de una mitad → calcular la otra → join — es el idioma canónico de fork/join. El truco de "calcular una mitad aquí en lugar de hacer fork de ambas" evita crear una tarea innecesaria y supone una pequeña pero real mejora.

El umbral importa

La decisión más importante: cuándo dejar de dividir. Un umbral demasiado pequeño genera miles de tareas para fragmentos triviales — la sobrecarga domina el trabajo. Demasiado grande y no puedes usar todos los núcleos — muchos workers permanecen inactivos mientras uno procesa un fragmento grande.

Reglas generales:

El cuerpo de la tarea debería tardar al menos 10 microsegundos. Por debajo de eso, la sobrecarga de gestión de tareas es comparable al trabajo.
Haz que el umbral sea una constante que puedas ajustar. 100, 1000, 10000 son valores típicos para arrays primitivos; el número correcto depende del coste por elemento.
Para entradas muy pequeñas, recurre a una implementación puramente serie. La sobrecarga de dividir y hacer fork se desperdicia en entradas que caben en caché.

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

Las tres operaciones sobre un RecursiveTask:

Método	Comportamiento
`fork()`	Programa la tarea en el pool actual; retorna inmediatamente
`join()`	Espera la tarea y retorna su resultado (o lanza lo que lanzó)
`invoke()`	Combinación de `fork` + `join` para este hilo — síncrono
`compute()`	Ejecuta el cuerpo directamente en el hilo actual (sin fork)

En el patrón anterior, left.fork(); right.compute(); left.join(); hace lo correcto — hace fork de una mitad en otro worker, ejecuta la otra mitad aquí, luego espera a que el fork termine.

No debes escribir left.fork(); right.fork(); left.join(); right.join();. El lado derecho se hace fork y el worker actual espera — no hay ningún hilo de ejecución disponible para realmente ejecutar right hasta que el worker llegue al join. La combinación desperdicia el turno del worker actual.

El pool común

ForkJoinPool.commonPool() es un pool compartido en toda la JVM, dimensionado a Runtime.getRuntime().availableProcessors() - 1 por defecto. Alimenta:

Stream.parallelStream()
CompletableFuture.supplyAsync(supplier) (la sobrecarga sin executor)
Arrays.parallelSort()

Puedes configurar el tamaño del pool común mediante la propiedad del sistema java.util.concurrent.ForkJoinPool.common.parallelism al iniciar la JVM. No debes usar el pool común para I/O — una única llamada bloqueante ocupa un worker que comparte toda la JVM.

Work-stealing en imágenes

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 cola de doble extremo es el corazón del diseño: los workers empujan y sacan del mismo extremo (LIFO — localidad de referencia para los hits de caché), los ladrones toman del otro extremo (FIFO — mínima contención con el propietario). Por eso fork/join rinde tan bien: los workers raramente compiten por las estructuras de datos de los demás, incluso bajo carga elevada.

Un ejemplo práctico: suma paralela vs. serie

El programa siguiente suma un array de 10 millones de elementos de dos formas — bucle serie y recursión fork/join — e imprime el tiempo de reloj para cada uno.

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

Lo que hay que extraer de la ejecución:

La versión fork/join fue varias veces más rápida que el bucle serie. En una máquina de N núcleos, el límite superior es aproximadamente N× — el número real fue menor porque la JVM, el GC y otros hilos de la JVM también demandaban CPU, y el trabajo del umbral no está perfectamente equilibrado. Aun así, una aceleración sustancial con unas pocas líneas de código recursivo.
Ambas sumas eran iguales. Esa es la comprobación de corrección de partición y combinación: cada hoja sumó su porción sin solapamiento; el paso de combinación (l + r) los añadió; sin doble conteo ni carreras de datos porque cada hoja escribió en su propia variable local.
La variante SumTiny con umbral 10 fue más lenta que el bucle serie. Con 10M de elementos divididos hasta fragmentos de 10, se crean unas 2M de tareas — y la sobrecarga de gestión de tareas supera ampliamente el trabajo de suma real. El umbral es un parámetro real; mídelo con entrada representativa.
El patrón left.fork(); long r = right.compute(); long l = left.join(); usó una tarea menos que fork(); fork(); join(); join();. El worker actual tiene tiempo libre durante el compute() — usarlo para una de las mitades ahorra una asignación de tarea completa. Esa es la pequeña pero acumulativa mejora en muchas cargas de trabajo reales.
ForkJoinPool.commonPool() fue el executor utilizado en esta demostración. Para una ejecución puntual, el pool común es suficiente. Para un programa de larga duración que mezcla trabajo fork/join con llamadas parallel de stream y futuros asíncronos, dale al trabajo fork/join intensivo su propio pool — el pool común está pensado para ráfagas cortas, no para cómputo intensivo continuo.

Qué viene a continuación

El siguiente capítulo, Java Concurrent Collections, cubre las estructuras de datos diseñadas para ser accedidas por múltiples hilos simultáneamente — ConcurrentHashMap, CopyOnWriteArrayList, BlockingQueue y el resto de java.util.concurrent.

Práctica

Dentro de `compute()` de un `RecursiveTask`, tienes dos subtareas `left` y `right`. ¿Cuál es el patrón de llamada canónico del idioma fork/join?

`left.fork(); long r = right.compute(); long l = left.join();` — hacer fork de una mitad, calcular la otra en el worker actual, hacer join del fork`left.fork(); right.fork(); long l = left.join(); long r = right.join();` — hacer fork de ambas y join de ambas`left.invoke(); right.invoke();` — invoke las ejecuta sincrónicamente en paralelo de forma automática`left.start(); right.start(); left.join(); right.join();` — igual que la API de `Thread` simple