1. ForkJoinPool: che cos’è e a cosa serve
ForkJoinPool è uno speciale pool di thread che implementa l’approccio “divide et impera” (divide and conquer). Il suo compito è parallelizzare in modo efficiente il lavoro quando un’attività grande può essere suddivisa in una serie di sottoattività indipendenti, eseguite in parallelo e poi confluire i risultati.
- Fork (dividere) — l’attività viene suddivisa in sottoattività.
- Join (unire) — i risultati delle sottoattività vengono aggregati nel risultato finale.
ForkJoinPool è il cuore dei parallel stream in Java: quando scrivi list.parallelStream(), dietro le quinte viene usato proprio lui. Ma puoi utilizzarlo anche direttamente, ottenendo un maggiore controllo.
Quando ForkJoinPool è particolarmente utile
ForkJoinPool dà il meglio con attività facilmente scomponibili in parti indipendenti: ad esempio l’elaborazione di array molto grandi, dove ogni frammento viene trattato separatamente e poi i risultati vengono uniti.
- L’attività si può dividere facilmente in sottoattività indipendenti: ordinamento, ricerca, somma.
- Le sottoattività sono all’incirca della stessa dimensione e non dipendono l’una dall’altra.
- Serve sfruttare tutti i core della CPU per massimizzare la velocità.
+---------------------+
| Grande task |
+---------------------+
|
v
+---------+---------+
| Sotto-task 1 |
| Sotto-task 2 |
| ... |
+-------------------+
|
v
+---------+---------+
| Risultati |
+-------------------+
È così che funziona “divide et impera”: si divide — si calcola in parallelo — si unisce.
2. RecursiveTask e RecursiveAction: due facce della stessa medaglia
In ForkJoinPool i task sono modellati tramite classi speciali capaci di suddividersi in sottoattività e di unire i risultati. RecursiveTask<T> restituisce un risultato, mentre RecursiveAction no. Nella pratica si usa più spesso RecursiveTask per, ad esempio, restituire una somma, un massimo o un conteggio.
Per creare un task di questo tipo, si estende la classe e si implementa il metodo compute(). Al suo interno si definisce la logica: se il task è piccolo — lo risolviamo subito; se è grande — lo dividiamo in sottoattività, le avviamo in parallelo tramite fork() e uniamo i risultati con join(). Così nasce un parallelismo ricorsivo naturale.
3. Sintassi ed esempio: calcolo parallelo della somma di un array
Supponiamo di avere un grande array di numeri e di voler calcolare rapidamente la somma di tutti gli elementi.
Passo 1. La classe del task
import java.util.concurrent.RecursiveTask;
public class ArraySumTask extends RecursiveTask<Long> {
private static final int THRESHOLD = 1_000; // Soglia per suddividere il task
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() {
// Se il task è piccolo — calcoliamo direttamente
if (end - start <= THRESHOLD) {
long sum = 0;
for (int i = start; i < end; i++) {
sum += array[i];
}
return sum;
} else {
// Suddividiamo il task in due sotto-task
int mid = (start + end) / 2;
ArraySumTask leftTask = new ArraySumTask(array, start, mid);
ArraySumTask rightTask = new ArraySumTask(array, mid, end);
// Avviamo i sotto-task in parallelo
leftTask.fork(); // Asincrono
long rightResult = rightTask.compute(); // Sincrono
long leftResult = leftTask.join(); // Attendiamo il completamento di leftTask
// Uniamo il risultato
return leftResult + rightResult;
}
}
}
- Se il task è piccolo (al di sotto della soglia THRESHOLD) — sommiamo con un ciclo tradizionale.
- Se è grande — lo dividiamo in due: uno lo avviamo in modo asincrono con fork(), l’altro lo calcoliamo in modo sincrono con compute(), poi uniamo con join().
Passo 2. Avvio del task tramite 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; // Per semplicità — la somma deve essere uguale alla lunghezza dell'array
}
ForkJoinPool pool = new ForkJoinPool(); // Per impostazione predefinita — pari al numero di core
ArraySumTask task = new ArraySumTask(numbers, 0, numbers.length);
long result = pool.invoke(task); // Avvio del task
System.out.println("Somma degli elementi dell'array: " + result);
}
}
Come funziona?
- ForkJoinPool decide autonomamente quanti thread usare (di solito — pari al numero di core).
- Il task viene automaticamente suddiviso in sottoattività, ciascuna delle quali può essere eseguita su un core separato.
- Le prestazioni sono in genere superiori rispetto al codice sequenziale (soprattutto su grandi moli di dati e sistemi multicore).
4. Come funziona ForkJoinPool: un po’ “sotto il cofano”
Work-stealing (furto di lavoro)
ForkJoinPool implementa il “furto di lavoro”: se un thread esaurisce i task, “ruba” lavoro a un altro thread. Questo garantisce un bilanciamento del carico efficace e l’utilizzo di tutti i core.
Algoritmo di base
- L’attività principale viene suddivisa in sottoattività.
- Le sottoattività vengono poste in code specializzate.
- I thread prelevano i task dalle proprie code e, se sono vuote, “cercano” lavoro presso i vicini.
- Al termine, i risultati vengono uniti.
Schema di funzionamento
flowchart TD
A[Attività principale] --> B1[Sottoattività 1]
A --> B2[Sottoattività 2]
B1 --> C1[Attività piccola 1]
B1 --> C2[Attività piccola 2]
B2 --> C3[Attività piccola 3]
B2 --> C4[Attività piccola 4]
C1 --> D[Aggregazione dei risultati]
C2 --> D
C3 --> D
C4 --> D
5. RecursiveAction — se non è necessario restituire un risultato
Se serve semplicemente eseguire qualcosa in parallelo senza restituire un risultato, usa RecursiveAction. Esempi tipici — parallelizzare il riempimento di un array, la stampa, l’ordinamento “in place” ecc.
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. Pratica: ricerca parallela del valore massimo in 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);
}
}
}
Avvio:
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("Valore massimo: " + max);
}
}
7. Vantaggi e limiti di ForkJoinPool
Vantaggi
- Bilanciamento automatico del carico. Il work-stealing consente di usare in modo efficiente tutti i core.
- Comodità. Non serve creare e gestire i thread manualmente.
- Alte prestazioni. Soprattutto su attività grandi e sistemi multicore.
- Flessibilità. Puoi suddividere i task in quante parti desideri.
Limiti
- Forte interdipendenza tra sottoattività. Se le sottoattività si aspettano spesso a vicenda, il vantaggio diminuisce.
- Task troppo piccoli. L’overhead di suddivisione/sincronizzazione può “mangiare” il beneficio del parallelismo.
- Effetti collaterali. Non si possono modificare variabili condivise senza sincronizzazione — si ottengono race condition.
- Applicabilità. È adatto per attività scomponibili in parti indipendenti.
8. Errori tipici nell’uso di ForkJoinPool e RecursiveTask
Errore n. 1: Suddivisione troppo fine del task. Se la soglia (THRESHOLD) è troppo piccola, si creano moltissimi task minuscoli — i costi di creazione e sincronizzazione supereranno il guadagno del parallelismo. Sperimenta con la soglia: spesso i valori ottimali sono nell’ordine delle migliaia o decine di migliaia di elementi.
Errore n. 2: Uso di variabili condivise e mutabili. Se le sottoattività scrivono su una variabile condivisa senza sincronizzazione — avrai corse critiche (race condition). Restituisci il risultato tramite compute() e unisci solo con join().
Errore n. 3: Uso scorretto di fork/join. Se dimentichi di chiamare fork() o join() — la sottoattività non verrà avviata in parallelo o il risultato andrà “perso”. Fai attenzione all’ordine delle chiamate.
Errore n. 4: Avviare ForkJoinTask al di fuori di ForkJoinPool. Se chiami semplicemente compute() sul task, esso verrà eseguito nel thread corrente, senza parallelismo. Per la vera magia usa pool.invoke() o pool.submit().
Errore n. 5: Ignorare le eccezioni. Se nel task si verifica un’eccezione, emergerà alla chiamata di join() o invoke(). Non dimenticare di gestire gli errori.
Errore n. 6: Usare ForkJoinPool per task bloccanti. ForkJoinPool non è adatto a task che si bloccano spesso (attendono I/O ecc.). In tali casi è meglio utilizzare ExecutorService.
GO TO FULL VERSION