Parallel Stream in Java: prestazioni ed errori

I Parallel Stream in Java permettono di elaborare gli elementi di una sorgente dati utilizzando più thread, senza dover suddividere manualmente il lavoro, creare task o coordinare direttamente i risultati.

È sufficiente sostituire stream() con parallelStream(), oppure richiamare il metodo parallel(), per chiedere alla piattaforma Java di eseguire la pipeline in modalità parallela.

La semplicità dell’API può però essere ingannevole. Un Parallel Stream non è automaticamente più veloce di uno Stream sequenziale e, se utilizzato nel contesto sbagliato, può aumentare il tempo di esecuzione, introdurre contesa tra thread o generare risultati non corretti.

Cosa sono i Parallel Stream in Java

Uno Stream rappresenta una sequenza di elementi sulla quale è possibile costruire una pipeline di operazioni aggregate.

Una pipeline può comprendere:

• una sorgente, come una lista, un array o un intervallo numerico;
• operazioni intermedie come filter(), map() e sorted();
• un’operazione terminale come collect(), sum(), reduce() o forEach().

In uno Stream sequenziale gli elementi vengono normalmente elaborati utilizzando il thread chiamante. In un Parallel Stream, invece, la sorgente viene suddivisa in più parti che possono essere elaborate contemporaneamente.

Il modello seguito può essere riassunto in quattro passaggi:

1. la sorgente viene suddivisa in parti più piccole;
2. le parti vengono assegnate a diversi task;
3. i task vengono elaborati da più thread;
4. i risultati parziali vengono combinati.

Differenza tra Stream sequenziale e Parallel Stream

Consideriamo una lista di numeri interi:

import java.util.List;
import java.util.stream.IntStream;

List<Integer> numeri = IntStream.rangeClosed(1, 10)
        .boxed()
        .toList();

Esecuzione sequenziale

numeri.stream()
        .forEach(numero -> System.out.printf(
                "%d - %s%n",
                numero,
                Thread.currentThread().getName()
        ));

Un possibile risultato è:

1 - main
2 - main
3 - main
4 - main
5 - main
6 - main
7 - main
8 - main
9 - main
10 - main

Tutta la pipeline viene eseguita dal thread main.

Esecuzione parallela

numeri.parallelStream()
        .forEach(numero -> System.out.printf(
                "%d - %s%n",
                numero,
                Thread.currentThread().getName()
        ));

L’output può cambiare a ogni esecuzione:

7 - main
3 - ForkJoinPool.commonPool-worker-1
9 - ForkJoinPool.commonPool-worker-2
6 - main
2 - ForkJoinPool.commonPool-worker-3
10 - ForkJoinPool.commonPool-worker-2
4 - ForkJoinPool.commonPool-worker-1
1 - ForkJoinPool.commonPool-worker-3
8 - main
5 - ForkJoinPool.commonPool-worker-1

Il thread chiamante può partecipare all’elaborazione insieme ai worker del pool. Non è garantito che ogni elemento venga elaborato da un thread differente.

Come creare un Parallel Stream

Java offre principalmente due modalità.

1. Collection.parallelStream()

List<String> nomi = List.of(
        "Anna",
        "Marco",
        "Paolo",
        "Giulia"
);

long totale = nomi.parallelStream()
        .filter(nome -> nome.length() > 4)
        .count();

2. BaseStream.parallel()

long totale = nomi.stream()
        .parallel()
        .filter(nome -> nome.length() > 4)
        .count();

I due esempi producono una pipeline parallela equivalente.

Verificare la modalità dello Stream

boolean parallelo = nomi.parallelStream()
        .isParallel();

System.out.println(parallelo); // true

Tornare alla modalità sequenziale

long risultato = nomi.parallelStream()
        .filter(nome -> nome.length() > 3)
        .sequential()
        .count();

La modalità utilizzata dall’intera pipeline è quella attiva quando viene eseguita l’operazione terminale.

Come funziona ForkJoinPool

I Parallel Stream utilizzano normalmente il ForkJoinPool.commonPool(), un pool condiviso all’interno della JVM.

Il framework Fork/Join applica una strategia chiamata work stealing. Quando un worker termina i propri task, può recuperare lavoro dalle code degli altri worker anziché rimanere inattivo.

Questa strategia è adatta a elaborazioni che possono essere suddivise in sottoproblemi indipendenti e successivamente ricomposte.

Visualizzare il livello di parallelismo

import java.util.concurrent.ForkJoinPool;

int parallelismo = ForkJoinPool.getCommonPoolParallelism();

System.out.println(
        "Parallelismo common pool: " + parallelismo
);

Il livello di parallelismo dipende dall’ambiente di esecuzione, dalla configurazione della JVM e dalle risorse disponibili.

Visualizzare lo stato del pool

ForkJoinPool pool = ForkJoinPool.commonPool();

System.out.println("Parallelismo: "
        + pool.getParallelism());

System.out.println("Thread attivi: "
        + pool.getActiveThreadCount());

System.out.println("Task accodati: "
        + pool.getQueuedTaskCount());

System.out.println("Steal count: "
        + pool.getStealCount());

Ordine degli elementi: forEach e forEachOrdered

Una List possiede un ordine di incontro, chiamato encounter order. Tuttavia, l’operazione terminale forEach() non garantisce che l’azione venga eseguita rispettando tale ordine quando lo Stream è parallelo.

Ordine non garantito

List.of(1, 2, 3, 4, 5)
        .parallelStream()
        .forEach(System.out::println);

Un possibile risultato è:

3
5
4
1
2

Preservare l’ordine con forEachOrdered()

List.of(1, 2, 3, 4, 5)
        .parallelStream()
        .forEachOrdered(System.out::println);

Il risultato rispetterà l’ordine originale:

1
2
3
4
5

Preservare l’ordine può però introdurre coordinamento e ridurre il vantaggio del parallelismo.

Rimuovere il vincolo di ordinamento

Quando l’ordine non è importante, il metodo unordered() può consentire alla pipeline parallela di lavorare con meno vincoli:

Set<Integer> valori = numeri.parallelStream()
        .unordered()
        .filter(numero -> numero % 2 == 0)
        .collect(java.util.stream.Collectors.toSet());

Questa ottimizzazione può essere utile soprattutto con operazioni come distinct(), limit() o alcune riduzioni raggruppate.

Stato condiviso ed effetti collaterali

Le funzioni passate a map(), filter() e alle altre operazioni dello Stream dovrebbero essere:

• non interferenti: non devono modificare la sorgente dello Stream;
• stateless: non devono dipendere da uno stato condiviso modificabile;
• thread-safe: non devono introdurre corse critiche tra thread.

Esempio errato: modificare un ArrayList condiviso

import java.util.ArrayList;
import java.util.List;

List<Integer> risultati = new ArrayList<>();

numeri.parallelStream()
        .filter(numero -> numero % 2 == 0)
        .forEach(risultati::add);

ArrayList non è thread-safe. Più worker possono tentare di modificarlo nello stesso momento, producendo risultati mancanti, incoerenti o eccezioni.

Soluzione corretta: utilizzare una riduzione

List<Integer> risultati = numeri.parallelStream()
        .filter(numero -> numero % 2 == 0)
        .toList();

La pipeline restituisce il risultato senza modificare uno stato esterno condiviso.

Una lista sincronizzata risolve il problema?

List<Integer> risultati =
        java.util.Collections.synchronizedList(
                new ArrayList<>()
        );

La sincronizzazione può evitare alcune corse critiche, ma introduce contesa. Diversi thread devono attendere per modificare la stessa struttura, riducendo o annullando il beneficio del parallelismo.

Riduzioni corrette con reduce()

Una riduzione combina più elementi per ottenere un singolo risultato, come una somma, un prodotto, un minimo o un massimo.

Somma parallela

long somma = java.util.stream.LongStream
        .rangeClosed(1, 10_000_000)
        .parallel()
        .sum();

Riduzione esplicita

int somma = numeri.parallelStream()
        .reduce(
                0,
                Integer::sum
        );

Per funzionare correttamente in parallelo, l’operazione di combinazione deve essere associativa.

In altre parole, il raggruppamento degli elementi non deve modificare il risultato:

(a + b) + c = a + (b + c)

Identità neutra

Il valore iniziale deve essere neutro rispetto all’operazione. Per una somma, l’identità corretta è zero:

int risultato = numeri.parallelStream()
        .reduce(0, Integer::sum);

Esempio scorretto

int risultato = numeri.parallelStream()
        .reduce(5, Integer::sum);

Il valore 5 potrebbe essere applicato a più risultati parziali. Non è quindi un’identità neutra e può produrre un risultato differente rispetto all’esecuzione sequenziale.

Operazione non associativa

La sottrazione non è una buona operazione di riduzione parallela:

int risultato = numeri.parallelStream()
        .reduce(0, (a, b) -> a - b);

Il risultato dipende da come gli elementi vengono suddivisi e ricombinati.

Collector e raccolte concorrenti

Il metodo collect() permette di eseguire una riduzione mutabile, costruendo una lista, una mappa o un altro contenitore.

Raccogliere gli elementi in una lista

List<String> risultati = nomi.parallelStream()
        .filter(nome -> nome.startsWith("A"))
        .map(String::toUpperCase)
        .toList();

Non è necessario utilizzare manualmente una lista sincronizzata. È il sistema di riduzione a creare e combinare correttamente i contenitori parziali.

Raggruppamento tradizionale

import static java.util.stream.Collectors.groupingBy;

Map<Integer, List<String>> nomiPerLunghezza =
        nomi.parallelStream()
                .collect(groupingBy(String::length));

groupingBy() non restituisce un collector concorrente. In una pipeline parallela, le mappe parziali devono essere combinate.

Utilizzare groupingByConcurrent()

import java.util.concurrent.ConcurrentMap;

import static java.util.stream.Collectors.groupingByConcurrent;

ConcurrentMap<Integer, List<String>> nomiPerLunghezza =
        nomi.parallelStream()
                .unordered()
                .collect(
                        groupingByConcurrent(
                                String::length
                        )
                );

Quando l’ordine non è necessario, groupingByConcurrent() può evitare parte del costo di fusione delle mappe parziali.

Contare le occorrenze

import java.util.concurrent.ConcurrentMap;
import java.util.function.Function;

import static java.util.stream.Collectors.counting;
import static java.util.stream.Collectors.groupingByConcurrent;

List<String> parole = List.of(
        "java",
        "stream",
        "java",
        "parallel",
        "stream",
        "java"
);

ConcurrentMap<String, Long> frequenze =
        parole.parallelStream()
                .unordered()
                .collect(
                        groupingByConcurrent(
                                Function.identity(),
                                counting()
                        )
                );

Quando un Parallel Stream è realmente più veloce

Il tempo richiesto da una pipeline parallela può essere rappresentato in modo semplificato come:

Il parallelismo è vantaggioso soltanto quando il risparmio ottenuto elaborando più parti contemporaneamente supera i costi aggiuntivi.

Fattori favorevoli

• grande quantità di elementi;
• operazione costosa per ogni elemento;
• lavoro principalmente CPU-bound;
• elementi indipendenti tra loro;
• sorgente facilmente suddivisibile;
• riduzione associativa ed efficiente;
• assenza di stato condiviso;
• ordine non necessario.

Fattori sfavorevoli

• pochi elementi;
• operazioni molto semplici;
• attività bloccanti o di rete;
• uso intenso di sincronizzazione;
• necessità di preservare rigorosamente l’ordine;
• sorgente difficile da suddividere;
• pool già utilizzato da altre attività;
• numero limitato di core disponibili.

Esempio con elaborazione CPU-bound

import java.util.stream.LongStream;

long totalePrimi = LongStream.rangeClosed(
                2,
                5_000_000
        )
        .parallel()
        .filter(ParallelExample::isPrimo)
        .count();

private static boolean isPrimo(long numero) {

    if (numero < 2) {
        return false;
    }

    long limite = (long) Math.sqrt(numero);

    for (long divisore = 2;
         divisore <= limite;
         divisore++) {

        if (numero % divisore == 0) {
            return false;
        }
    }

    return true;
}

In questo caso ogni elemento richiede una quantità di lavoro maggiore rispetto a una semplice somma o stampa su console. Il parallelismo ha quindi più possibilità di compensare il costo di suddivisione e coordinamento.

Quando utilizzare parallelStream()

Un Parallel Stream può essere una buona scelta quando sono soddisfatte contemporaneamente diverse condizioni.

Esempio:

double media = prodotti.parallelStream()
        .filter(Prodotto::isDisponibile)
        .mapToDouble(Prodotto::getPrezzo)
        .average()
        .orElse(0.0);

La pipeline è adatta al parallelismo quando i metodi utilizzati non modificano lo stato dei prodotti e il dataset è sufficientemente grande.

Quando evitare i Parallel Stream

1. Collezioni molto piccole

List.of(1, 2, 3)
        .parallelStream()
        .map(numero -> numero * 2)
        .toList();

Per tre semplici moltiplicazioni, il costo del parallelismo è superiore al lavoro utile.

2. Operazioni estremamente economiche

long somma = numeri.parallelStream()
        .mapToLong(Integer::longValue)
        .sum();

La somma parallela può essere utile su dataset molto grandi, ma non deve essere considerata automaticamente migliore.

3. Stato condiviso

L’utilizzo di contatori mutabili, liste esterne o servizi non thread-safe può produrre risultati errati.

4. Ordine obbligatorio

Operazioni come forEachOrdered(), limit(), skip() o distinct() possono richiedere coordinamento e buffering quando devono rispettare l’ordine.

5. Applicazioni server condivise

In un server web possono essere elaborate contemporaneamente numerose richieste. Ogni richiesta che avvia un Parallel Stream può competere per le stesse risorse del common pool.

6. Operazioni bloccanti

Chiamate HTTP, query al database, lettura di file remoti o attese esplicite possono bloccare i worker del pool.

Parallel Stream e operazioni bloccanti

Il ForkJoinPool è progettato principalmente per task che si suddividono in sottotask e consumano CPU. Non garantisce una compensazione efficace quando molti worker rimangono bloccati su operazioni di input/output o sincronizzazione non gestita.

Esempio da evitare

List<String> contenuti = urls.parallelStream()
        .map(httpClient::scaricaPagina)
        .toList();

Ogni richiesta di rete può rimanere in attesa. Se i worker del common pool vengono occupati, altre elaborazioni parallele della JVM possono subire rallentamenti.

Alternativa moderna con Virtual Thread

Da Java 21, per molte attività bloccanti è possibile utilizzare un executor basato su Virtual Thread:

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

List<String> contenuti = new ArrayList<>();

try (ExecutorService executor =
             Executors.newVirtualThreadPerTaskExecutor()) {

    List<Future<String>> futures = urls.stream()
            .map(url -> executor.submit(
                    () -> httpClient.scaricaPagina(url)
            ))
            .toList();

    for (Future<String> future : futures) {
        contenuti.add(future.get());
    }
}

I Virtual Thread non rendono più veloce un calcolo CPU-bound, ma permettono di gestire un elevato numero di attività che trascorrono molto tempo in attesa.

Gestione delle eccezioni nei Parallel Stream

Le interfacce funzionali standard, come Function e Consumer, non dichiarano checked exception. Inoltre, un’eccezione non gestita in un elemento può far fallire l’intera operazione terminale.

Gestione locale dell’errore

List<Integer> valori = testi.parallelStream()
        .map(testo -> {
            try {
                return Integer.parseInt(testo);
            } catch (NumberFormatException exception) {
                return null;
            }
        })
        .filter(java.util.Objects::nonNull)
        .toList();

Questo approccio può essere sufficiente quando gli elementi non validi devono semplicemente essere ignorati. Tuttavia, si perdono le informazioni sull’errore.

Conservare sia risultati sia errori

record Esito<T>(
        T valore,
        Exception errore
) {

    static <T> Esito<T> successo(T valore) {
        return new Esito<>(valore, null);
    }

    static <T> Esito<T> fallimento(Exception errore) {
        return new Esito<>(null, errore);
    }

    boolean completato() {
        return errore == null;
    }
}

List<Esito<Integer>> esiti = testi.parallelStream()
        .map(testo -> {
            try {
                return Esito.successo(
                        Integer.parseInt(testo)
                );
            } catch (NumberFormatException exception) {
                return Esito.<Integer>fallimento(
                        exception
                );
            }
        })
        .toList();

Ogni elemento produce un risultato esplicito e l’intera elaborazione non viene interrotta dal primo valore non valido.

Misurare le prestazioni con JMH

Il confronto tra Stream sequenziale e parallelo non dovrebbe basarsi su una singola esecuzione misurata con System.currentTimeMillis().

La JVM applica compilazione JIT, ottimizzazioni, garbage collection e warm-up. Una misurazione improvvisata può quindi essere fuorviante.

JMH, sviluppato nell’ambito OpenJDK, è lo strumento di riferimento per creare microbenchmark sulla JVM.

Esempio di benchmark

import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
import org.openjdk.jmh.annotations.Fork;
import org.openjdk.jmh.annotations.Measurement;
import org.openjdk.jmh.annotations.Mode;
import org.openjdk.jmh.annotations.OutputTimeUnit;
import org.openjdk.jmh.annotations.Scope;
import org.openjdk.jmh.annotations.State;
import org.openjdk.jmh.annotations.Warmup;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.LongStream;

@State(Scope.Benchmark)
@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
@Warmup(iterations = 5)
@Measurement(iterations = 10)
@Fork(2)
public class ParallelStreamBenchmark {

    private final List<Long> numeri =
            LongStream.rangeClosed(1, 2_000_000)
                    .boxed()
                    .toList();

    @Benchmark
    public long sequenziale() {
        return numeri.stream()
                .mapToLong(this::calcolo)
                .sum();
    }

    @Benchmark
    public long parallelo() {
        return numeri.parallelStream()
                .mapToLong(this::calcolo)
                .sum();
    }

    private long calcolo(long valore) {

        long risultato = valore;

        for (int i = 0; i < 100; i++) {
            risultato = Long.rotateLeft(
                    risultato * 31,
                    3
            );
        }

        return risultato;
    }
}

Cosa deve rappresentare il benchmark

• dimensione reale del dataset;
• costo effettivo della funzione;
• hardware simile alla produzione;
• versione reale della JVM;
• carico concorrente dell’applicazione;
• vincoli di ordinamento;
• eventuali allocazioni e garbage collection.

Diagnostica e monitoraggio

Durante lo sviluppo è possibile osservare i thread utilizzati dalla pipeline:

numeri.parallelStream()
        .map(numero -> {
            System.out.printf(
                    "Valore %d elaborato da %s%n",
                    numero,
                    Thread.currentThread().getName()
            );

            return numero * 2;
        })
        .toList();

La stampa su console introduce però sincronizzazione e input/output. Non deve quindi essere utilizzata per misurare le prestazioni reali.

Informazioni sul sistema

int processori =
        Runtime.getRuntime().availableProcessors();

int parallelismo =
        ForkJoinPool.getCommonPoolParallelism();

System.out.println(
        "Processori disponibili: " + processori
);

System.out.println(
        "Parallelismo common pool: " + parallelismo
);

Osservare il pool

ForkJoinPool pool = ForkJoinPool.commonPool();

System.out.println(pool);

Il metodo toString() restituisce una rappresentazione informativa dello stato del pool, comprendente parallelismo, worker, task e stato di esecuzione.

Checklist prima di utilizzare un Parallel Stream

Se diverse risposte sono negative o incerte, uno Stream sequenziale è probabilmente la soluzione più sicura.

Domande frequenti sui Parallel Stream

parallelStream() crea un thread per ogni elemento?

No. Java suddivide la sorgente in task che vengono elaborati da un numero limitato di worker. Uno stesso worker può elaborare numerosi elementi.

Un Parallel Stream è sempre più veloce?

No. Il costo di suddivisione, scheduling, sincronizzazione e combinazione può renderlo più lento di uno Stream sequenziale.

Quale pool viene utilizzato?

Normalmente le operazioni parallele utilizzano il ForkJoinPool.commonPool(), condiviso all’interno della JVM.

Il thread main partecipa all’elaborazione?

Può partecipare. L’output degli esempi mostra spesso sia il thread chiamante sia i worker del common pool.

forEach() conserva l’ordine?

Non in una pipeline parallela. Per rispettare l’ordine di incontro è possibile utilizzare forEachOrdered(), accettando un possibile costo prestazionale.

Posso modificare una lista esterna dentro forEach()?

È sconsigliato. Una raccolta non thread-safe può produrre risultati errati, mentre una raccolta sincronizzata può introdurre contesa. È preferibile utilizzare collect() o toList().

Posso usare parallelStream() per chiamate HTTP?

Tecnicamente è possibile, ma generalmente non è la scelta migliore. Le operazioni bloccanti possono occupare i worker del common pool. Un executor esplicito o i Virtual Thread offrono un controllo più adatto.

parallelStream() è adatto in Spring Boot?

Può essere utilizzato, ma occorre considerare che più richieste HTTP possono condividere contemporaneamente lo stesso common pool. Prima di introdurlo in un endpoint è necessario misurarne il comportamento sotto carico.

Come posso sapere se la pipeline è parallela?

Attraverso il metodo isParallel():

boolean parallelo = stream.isParallel();

Quali sorgenti si dividono meglio?

Array, ArrayList e intervalli numerici permettono generalmente una suddivisione efficiente. Strutture che richiedono una scansione per individuare il punto di divisione possono offrire risultati meno favorevoli.

Come decido tra Stream sequenziale e parallelo?

Si parte normalmente dalla versione sequenziale, si mantiene la pipeline priva di effetti collaterali e si confrontano le due modalità con JMH o con test prestazionali rappresentativi del carico reale.

Conclusioni

I Parallel Stream offrono una sintassi estremamente compatta per eseguire elaborazioni dati su più thread. La piattaforma Java si occupa della suddivisione della sorgente, della gestione dei task e della combinazione dei risultati.

Questa comodità non elimina però i costi e i rischi della programmazione parallela. Ordine, stato condiviso, sincronizzazione, operazioni bloccanti, dimensione del dataset e caratteristiche della sorgente possono cambiare completamente il risultato prestazionale.

La strategia più sicura consiste nel costruire inizialmente una pipeline sequenziale, mantenere le funzioni stateless, utilizzare riduzioni associative e attivare il parallelismo soltanto dopo aver misurato un miglioramento concreto.

Documentazione tecnica:
Java Stream API,
ForkJoinPool,
Collectors,
OpenJDK JMH e
Virtual Thread.