L'API Stream distingue tra operazioni stateless (filter, map) e operazioni stateful (sorted, distinct, limit) in base al fatto che debbano elaborare l'intero input prima di produrre output. Durante l'esecuzione in parallelo, il framework partiziona i dati di origine tra più thread, ciascuno dei quali elabora un segmento in modo indipendente. Se lo Spliterator di origine riporta la caratteristica ORDERED, il framework assume che l'ordine di incontro (l'ordine in cui gli elementi appaiono nella sorgente) sia significativo e debba essere mantenuto durante tutta la pipeline.
Tuttavia, le operazioni stateful come distinct si basano su uno stato globale (un Set di elementi visti) per filtrare i duplicati. Senza una forza di ordine di incontro esplicita, i thread paralleli possono gareggiare per rivendicare gli elementi come la "prima" occorrenza, portando a una selezione arbitraria di quale duplicato sopravviva. Allo stesso modo, sorted richiede un'ordinamento globale, ma se lo stream è contrassegnato come unordered o la sorgente manca della caratteristica ORDERED, i risultati intermedi provenienti dai thread paralleli potrebbero essere uniti senza preservare la posizione. Questo può portare a diversi ordinamenti relativi di elementi uguali o, in casi degenerati, a una apparente non determinismo nella sequenza di output.
La soluzione sta nel rispettare il contratto dello Spliterator: se l'ordine di incontro è importante, la sorgente deve dichiarare ORDERED e la pipeline non deve invocare unordered() prima di un'operazione stateful. Per distinct, ciò garantisce che la "prima" occorrenza nell'ordine di incontro venga selezionata in modo deterministico elaborando i segmenti dello stream in ordine di sequenza, riducendo effettivamente il parallelismo per quella fase. Se l'ordine non è rilevante, la chiamata esplicita di unordered() consente al framework di ottimizzare selezionando duplicati arbitrari e fondendo risultati parziali senza sincronizzazione, migliorando le prestazioni a scapito della determinazione.
Un sistema di elaborazione della telemetria ha acquisito milioni di eventi sensoriali, ciascuno contrassegnato con un timestamp in nanosecondi e un ID sensore unico. Il requisito era quello di deduplicare gli eventi per ID sensore mantenendo l'evento cronologicamente primo per ciascun ID, quindi ordinare il resto per timestamp. L'implementazione iniziale ha utilizzato sensorReadings.parallelStream().distinct().sorted(), assumendo che la sorgente ArrayList mantenesse l'ordine di inserimento e che distinct avrebbe naturalmente preservato la prima occorrenza.
Il problema si è manifestato come fallimenti intermittenti nei test in cui l'evento "primo" per un dato ID sensore era casualmente la seconda o terza occorrenza nell'elenco originale quando eseguito su hardware multi-core. A seguito di un'indagine, il problema è stato ricondotto a distinct che veniva eseguito in parallelo senza forzare l'ordine di incontro; ogni thread elaborava un blocco dell'elenco e manteneva la propria "prima" occorrenza locale di ciascun ID. Quando il framework ha unito questi risultati parziali, l'ordinamento globale dei thread non era garantito, causando una selezione arbitraria tra le prime locali dei thread.
Sono state valutate tre soluzioni. Il primo approccio ha abbandonato completamente il parallelismo, tornando a uno stream sequenziale. Questo ha ripristinato un comportamento deterministico, assicurando che l'evento più precoce nell'elenco vincesse sempre. Tuttavia, ha aumentato la latenza di elaborazione del 400% durante i picchi di carico, violando gli SLA di throughput e richiedendo aggiornamenti hardware che non erano previsti nel budget.
Il secondo approccio ha inserito .unordered() prima di distinct, segnalando esplicitamente che qualsiasi duplicato era accettabile. Questo ha massimizzato il throughput consentendo ai thread di scartare duplicati arbitrari senza coordinamento. Sfortunatamente, questo ha violato il requisito aziendale di mantenere la lettura più precoce, rendendo l'approccio inaccettabile per la tracciabilità.
Il terzo approccio ha sfruttato un LinkedHashSet come collezionatore a valle tramite Collectors.toCollection(LinkedHashSet::new) all'interno di un'operazione di collect. Questo ha materializzato lo stream in un insieme ordinato, consentendo comunque la decomposizione parallela per le operazioni di filtro precedenti. Tuttavia, ciò ha richiesto di abbandonare l'operazione intermedia distinct e ha consumato significativamente più memoria per contenere l'intero set di lavoro prima della deduplicazione.
La soluzione scelta ha coinvolto la ristrutturazione della pipeline per separare le fasi ordinate e non ordinate. Il sistema ha prima applicato filtraggio e mapping stateless in parallelo, quindi è passato esplicitamente a uno stream sequenziale tramite .sequential() prima di invocare distinct e sorted. Questo approccio ibrido ha limitato il collo di bottiglia sequenziale solo alla parte terminale stateful, preservando il 70% del throughput parallelo mentre garantiva l'ordine di incontro.
Il risultato è stata una pipeline stabile e deterministica che ha identificato correttamente la prima occorrenza di ciascun evento sensoriale. Le velocità di elaborazione sono rimaste accettabili e il tasso di difetti è sceso a zero mentre la latenza è rimasta all'interno delle soglie operative.
Perché l'operazione terminale forEachOrdered comporta costi significativamente più elevati rispetto a forEach negli stream paralleli, e quando è strettamente necessaria?
forEach elabora gli elementi man mano che diventano disponibili dai thread paralleli senza coordinamento. Questo approccio massimizza il throughput ma produce potenzialmente output nell'ordine di arrivo dei thread. forEachOrdered, al contrario, deve ricostruire l'ordine di incontro originale, richiedendo al framework di bufferizzare i risultati e potenzialmente fermare i thread veloci ad aspettare quelli più lenti che possiedono elementi precedenti, creando un collo di bottiglia di sincronizzazione.
È strettamente necessaria solo quando gli effetti collaterali dell'elaborazione devono osservare l'ordine della sorgente. Esempi includono la scrittura su un output sensibile all'ordine come un file o un modello di elenco GUI. Per effetti collaterali non sensibili all'ordine come la registrazione o la somma in una collezione concorrente, si preferisce forEach.
Come il requisito dell'operazione reduce per una funzione accumulatrice associativa previene sottili condizioni di gara durante l'esecuzione parallela, e cosa succede se questo vincolo viene violato?
L'operazione reduce partiziona lo stream in segmenti, applica l'accumulatore a ciascun segmento in isolamento per produrre risultati parziali, e quindi combina questi risultati parziali utilizzando lo stesso accumulatore (o un combinatore separato). L'associatività assicura che ((a op b) op c) sia uguale a (a op (b op c)). Questa proprietà è necessaria perché il raggruppamento degli elementi in segmenti e l'ordine di combinazione dei risultati parziali è non deterministico e dipendente dall'implementazione.
Se l'operazione non è associativa (ad esempio, la concatenazione di stringhe con un delimitatore che varia a seconda della posizione), l'esecuzione parallela può raggruppare gli elementi in modo diverso rispetto all'esecuzione sequenziale. Questo produce risultati errati come delimitatori mescolati o somme matematicamente sbagliate per tipi di numero personalizzati non associativi.
Quale interazione specifica tra operazioni di short-circuiting come findFirst e stream infiniti fa sì che uno stream parallelo possa potenzialmente bloccarsi indefinitamente, mentre uno stream sequenziale terminerebbe immediatamente?
In uno stream sequenziale, findFirst può terminare non appena il predicato corrisponde, anche su uno stream infinito. In uno stream parallelo, il framework divide la sorgente in più segmenti elaborati da thread diversi. Se l'elemento corrispondente si trova in un segmento elaborato da un thread lento, findFirst deve attendere che quel thread completi il suo segmento (o trovi l'elemento) per garantire che non esista un elemento precedente in altri segmenti, poiché deve rispettare l'ordine di incontro.
Se lo stream è unordered o viene utilizzato findAny invece, l'operazione può terminare immediatamente su qualsiasi corrispondenza, consentendo al thread principale di annullare i compiti in attesa. I candidati spesso trascurano che findFirst su stream paralleli infiniti ordinati è effettivamente una barriera globale che può bloccarsi se i segmenti avanti rispetto alla corrispondenza sono infiniti o computazionalmente illimitati.