1. ForkJoinPool: qué es y para qué sirve
ForkJoinPool es un pool de hilos especializado que implementa el enfoque «divide y vencerás» (divide and conquer). Su objetivo es paralelizar el trabajo de la forma más eficiente posible cuando una tarea grande puede descomponerse en varias subtareas independientes, ejecutarlas en paralelo y luego combinar los resultados.
- Fork (dividir): la tarea se divide en subtareas.
- Join (unir): los resultados de las subtareas se combinan en el resultado final.
ForkJoinPool es el corazón de los streams paralelos en Java: cuando escribe list.parallelStream(), internamente se usa este pool. Pero también puede utilizarlo directamente para obtener más control.
Cuándo es especialmente útil ForkJoinPool
ForkJoinPool brilla en tareas que se descomponen fácilmente en partes independientes: por ejemplo, el procesamiento de arrays muy grandes, donde cada fragmento se procesa por separado y luego se combinan los resultados.
- La tarea se puede dividir fácilmente en subtareas independientes: ordenación, búsqueda, suma.
- Las subtareas son aproximadamente del mismo tamaño y no dependen entre sí.
- Se necesita aprovechar todos los núcleos del procesador para la máxima velocidad.
+---------------------+
| Tarea grande |
+---------------------+
|
v
+---------+---------+
| Subtarea 1 |
| Subtarea 2 |
| ... |
+-------------------+
|
v
+---------+---------+
| Resultados |
+-------------------+
Así funciona «divide y vencerás»: dividir — calcular en paralelo — unir.
2. RecursiveTask y RecursiveAction: dos caras de la misma moneda
En ForkJoinPool las tareas se modelan mediante clases especiales capaces de dividirse en subtareas y combinar resultados. RecursiveTask<T> devuelve un resultado, mientras que RecursiveAction no. En la práctica, se usa más RecursiveTask para, por ejemplo, devolver una suma, un máximo o un recuento.
Para crear una tarea de este tipo, heredamos e implementamos el método compute(). En él describimos la lógica: si la tarea es pequeña, la resolvemos inmediatamente; si es grande, la dividimos en subtareas, las lanzamos en paralelo con fork() y combinamos los resultados mediante join(). Así nace un paralelismo recursivo natural.
3. Sintaxis y ejemplo: cálculo paralelo de la suma de un array
Supongamos que tenemos un array grande de números y queremos calcular rápidamente la suma de todos sus elementos.
Paso 1. Clase de la tarea
import java.util.concurrent.RecursiveTask;
public class ArraySumTask extends RecursiveTask<Long> {
private static final int THRESHOLD = 1_000; // Umbral para dividir la tarea
private final int[] array;
private final int start, end;
public ArraySumTask(int[] array, int start, int end) {
this.array = array;
this.start = start;
this.end = end;
}
@Override
protected Long compute() {
// Si la tarea es pequeña, calculamos directamente
if (end - start <= THRESHOLD) {
long sum = 0;
for (int i = start; i < end; i++) {
sum += array[i];
}
return sum;
} else {
// Dividimos la tarea en dos subtareas
int mid = (start + end) / 2;
ArraySumTask leftTask = new ArraySumTask(array, start, mid);
ArraySumTask rightTask = new ArraySumTask(array, mid, end);
// Lanzamos las subtareas en paralelo
leftTask.fork(); // Asíncrono
long rightResult = rightTask.compute(); // Sincrónico
long leftResult = leftTask.join(); // Esperamos a que termine la izquierda
// Combinamos el resultado
return leftResult + rightResult;
}
}
}
- Si la tarea es pequeña (menor que el umbral THRESHOLD), calculamos la suma con un bucle normal.
- Si es grande, la dividimos en dos; una se lanza de forma asíncrona con fork(), la otra se calcula sincrónicamente con compute(), y luego se combinan mediante join().
Paso 2. Ejecución de la tarea mediante ForkJoinPool
import java.util.concurrent.ForkJoinPool;
public class ForkJoinSumDemo {
public static void main(String[] args) {
int[] numbers = new int[10_000_000];
for (int i = 0; i < numbers.length; i++) {
numbers[i] = 1; // Para simplificar, la suma debe ser igual a la longitud del array
}
ForkJoinPool pool = new ForkJoinPool(); // Por defecto, según el número de núcleos
ArraySumTask task = new ArraySumTask(numbers, 0, numbers.length);
long result = pool.invoke(task); // Inicio de la tarea
System.out.println("Suma de los elementos del array: " + result);
}
}
¿Cómo funciona?
- ForkJoinPool decide por sí mismo cuántos hilos usar (normalmente, según el número de núcleos).
- La tarea se divide automáticamente en subtareas, cada una de las cuales puede ejecutarse en un núcleo distinto.
- El rendimiento suele ser mayor que en el código secuencial (especialmente con datos grandes y en sistemas multinúcleo).
4. Cómo funciona ForkJoinPool: un poco «bajo el capó»
Work-stealing (robo de trabajo)
ForkJoinPool implementa el «robo de trabajo»: si un hilo se queda sin tareas, «roba» trabajo de otro hilo. Esto proporciona un balanceo de carga eficiente y mantiene ocupados todos los núcleos.
Algoritmo básico
- La tarea principal se divide en subtareas.
- Las subtareas se colocan en colas especializadas.
- Los hilos toman tareas de sus propias colas y, si se vacían, «buscan» trabajo en las colas vecinas.
- Cuando todo termina, se combinan los resultados.
Esquema de funcionamiento
flowchart TD
A[Tarea principal] --> B1[Subtarea 1]
A --> B2[Subtarea 2]
B1 --> C1[Tarea pequeña 1]
B1 --> C2[Tarea pequeña 2]
B2 --> C3[Tarea pequeña 3]
B2 --> C4[Tarea pequeña 4]
C1 --> D[Combinación de resultados]
C2 --> D
C3 --> D
C4 --> D
5. RecursiveAction: cuando no hace falta devolver un resultado
Si solo necesita ejecutar algo en paralelo y no devolver un resultado, use RecursiveAction. Ejemplos típicos: paralelizar el llenado de un array, impresión, ordenaciones in situ, etc.
import java.util.concurrent.RecursiveAction;
public class PrintTask extends RecursiveAction {
private static final int THRESHOLD = 100;
private final int[] array;
private final int start, end;
public PrintTask(int[] array, int start, int end) {
this.array = array;
this.start = start;
this.end = end;
}
@Override
protected void compute() {
if (end - start <= THRESHOLD) {
for (int i = start; i < end; i++) {
System.out.print(array[i] + " ");
}
} else {
int mid = (start + end) / 2;
invokeAll(
new PrintTask(array, start, mid),
new PrintTask(array, mid, end)
);
}
}
}
6. Práctica: búsqueda paralela del valor máximo en un array
import java.util.concurrent.RecursiveTask;
public class MaxFindTask extends RecursiveTask<Integer> {
private static final int THRESHOLD = 1000;
private final int[] array;
private final int start, end;
public MaxFindTask(int[] array, int start, int end) {
this.array = array;
this.start = start;
this.end = end;
}
@Override
protected Integer compute() {
if (end - start <= THRESHOLD) {
int max = array[start];
for (int i = start + 1; i < end; i++) {
if (array[i] > max) max = array[i];
}
return max;
} else {
int mid = (start + end) / 2;
MaxFindTask left = new MaxFindTask(array, start, mid);
MaxFindTask right = new MaxFindTask(array, mid, end);
left.fork();
int rightResult = right.compute();
int leftResult = left.join();
return Math.max(leftResult, rightResult);
}
}
}
Ejecución:
import java.util.concurrent.ForkJoinPool;
public class ForkJoinMaxDemo {
public static void main(String[] args) {
int[] array = new int[5_000_000];
for (int i = 0; i < array.length; i++) {
array[i] = (int)(Math.random() * 1_000_000);
}
ForkJoinPool pool = new ForkJoinPool();
MaxFindTask task = new MaxFindTask(array, 0, array.length);
int max = pool.invoke(task);
System.out.println("Valor máximo: " + max);
}
}
7. Ventajas y limitaciones de ForkJoinPool
Ventajas
- Balanceo de carga automático. El work-stealing permite utilizar eficientemente todos los núcleos.
- Comodidad. No es necesario crear y gestionar hilos manualmente.
- Alto rendimiento. Especialmente en tareas grandes y sistemas multinúcleo.
- Flexibilidad. Puede dividir las tareas en tantas partes como sea necesario.
Limitaciones
- Fuerte acoplamiento entre subtareas. Si las subtareas se esperan con frecuencia entre sí, la ganancia disminuye.
- Tareas demasiado pequeñas. La sobrecarga de división/sincronización puede «comerse» la ventaja.
- Efectos secundarios. No se deben modificar variables compartidas sin sincronización: obtendrá una race condition.
- Aplicabilidad. Es adecuado para tareas que se pueden dividir en partes independientes.
8. Errores típicos al trabajar con ForkJoinPool y RecursiveTask
Error n.º 1: división de la tarea demasiado fina. Si el umbral (THRESHOLD) es demasiado pequeño, habrá muchas tareas diminutas: el coste de crearlas y sincronizarlas superará la ganancia del paralelismo. Pruebe con el umbral: los valores óptimos suelen estar en miles o decenas de miles de elementos.
Error n.º 2: uso de variables compartidas mutables. Si las subtareas escriben en una variable compartida sin sincronización, obtendrá condiciones de carrera (race condition). Devuelva el resultado a través de compute() y combine solo en join().
Error n.º 3: uso incorrecto de fork/join. Si olvida llamar a fork() o join(), la subtarea no se ejecutará en paralelo o el resultado «se perderá». Vigile cuidadosamente el orden de las llamadas.
Error n.º 4: ejecutar una ForkJoinTask fuera de un ForkJoinPool. Si simplemente llama a compute() de la tarea, esta se ejecutará en el hilo actual, sin paralelismo. Para que ocurra la magia, use pool.invoke() o pool.submit().
Error n.º 5: ignorar las excepciones. Si en la tarea se produce una excepción, esta se manifestará al llamar a join() o invoke(). No olvide gestionar los errores.
Error n.º 6: usar ForkJoinPool para tareas con bloqueos. ForkJoinPool no es adecuado para tareas que se bloquean con frecuencia (esperan E/S (I/O), etc.). En esos casos, es mejor usar ExecutorService.
GO TO FULL VERSION