De Stream API maakt onderscheid tussen stateless operaties (filter, map) en stateful operaties (sorted, distinct, limit) op basis van of ze de volledige invoer moeten verwerken voordat ze uitvoer produceren. Bij parallelle uitvoering partitioneert het framework de brond gegevens over meerdere threads, die elke een segment onafhankelijk verwerken. Als de Spliterator de ORDERED eigenschap aangeeft, gaat het framework ervan uit dat de encounter order (de volgorde waarin elementen in de bron verschijnen) significant is en gedurende de pijplijn behouden moet blijven.
Echter, stateful operaties zoals distinct vertrouwen op een globale status (een Set van geziene elementen) om duplicaten te filteren. Zonder expliciete afdwinging van de encounter order kunnen parallelle threads strijden om elementen als de "eerste" verschijning te claimen, wat leidt tot arbitraire selectie van welk duplicaat overleeft. Evenzo vereist sorted een globale sortering, maar als de stream gemarkeerd is als unordered of de bron de ORDERED eigenschap mist, kunnen tussentijdse resultaten van parallelle threads samengevoegd worden zonder positiebehoud. Dit kan verschillende relatieve ordeningen van gelijke elementen opleveren of, in degenerate gevallen, schijnbare niet-determinisme in de uitvoervolgorde.
De oplossing ligt in het respecteren van het Spliterator-contract: als de encounter order belangrijk is, moet de bron ORDERED verklaren, en de pijplijn mag unordered() niet aanroepen vóór een stateful operatie. Voor distinct zorgt dit ervoor dat de "eerste" verschijning in encounter order deterministisch wordt geselecteerd door stream-segmenten in sequentiële volgorde te verwerken, waardoor de parallelisme voor die fase effectief wordt verminderd. Als de volgorde irrelevant is, stelt het expliciet aanroepen van unordered() het framework in staat om te optimaliseren door willekeurige duplicaten te selecteren en gedeeltelijke resultaten zonder synchronisatie samen te voegen, wat de prestaties verbetert ten koste van determinisme.
Een telemetrie-verwerkingssysteem verwerkte miljoenen sensor evenementen, elk gemarkeerd met een nanoseconde tijdstempel en een unieke sensor-ID. De vereiste was om evenementen te dedupliceren op sensor-ID, terwijl het chronologisch eerste evenement voor elke ID behouden bleef, en vervolgens de rest te sorteren op tijdstempel. De oorspronkelijke implementatie gebruikte sensorReadings.parallelStream().distinct().sorted(), in de veronderstelling dat de ArrayList-bron de invoervolgorde zou behouden en dat distinct van nature de eerste verschijning zou behouden.
Het probleem manifesteerde zich als intermitterende testfouten waarbij het "eerste" evenement voor een gegeven sensor-ID willekeurig de tweede of derde verschijning in de oorspronkelijke lijst bleek te zijn wanneer het op multicore-hardware werd uitgevoerd. Na onderzoek werd het probleem teruggevoerd naar distinct die parallel werd uitgevoerd zonder afdwinging van de encounter order; elke thread verwerkte een stuk van de lijst en behield zijn eigen lokale "eerste" ontmoeting van elke ID. Toen het framework deze gedeeltelijke resultaten samenvoegde, was de globale ordering van threads niet gegarandeerd, wat leidde tot arbitraire selectie tussen thread-lokale eersten.
Drie oplossingen werden geëvalueerd. De eerste aanpak verliet volledig het parallelisme en keerde terug naar een sequentiële stream. Dit herstelde het deterministisch gedrag, waarbij het vroegste evenement in de lijst altijd gewonnen had. Echter, het verhoogde de verwerkinglatentie met 400% onder piekbelasting, wat de doorvoersnelheden overtrad en hardware-upgrades vereiste die niet begroot waren.
De tweede aanpak voegde .unordered() toe vóór distinct, wat expliciet aangaf dat elke duplicaat acceptabel was. Dit maximaliseerde de doorvoer door threads in staat te stellen willekeurige duplicaten zonder coördinatie te verwerpen. Helaas schond dit de zakelijke vereiste om het vroegste lezen te behouden, waardoor de aanpak onacceptabel werd voor het auditspoor.
De derde aanpak gebruikte een LinkedHashSet als een downstream verzamelaar via Collectors.toCollection(LinkedHashSet::new) binnen een collect operatie. Dit maakte de stream in een geordende set, terwijl het nog steeds parallelle decompositie voor voorgaande filteroperaties mogelijk maakte. Echter, dit vereiste het opgeven van de tussenliggende distinct operatie en verbruikte aanzienlijk meer geheugen om de volledige werkset vast te houden voordat de duplicatie plaatsvond.
De gekozen oplossing hield in dat de pijplijn werd herstructureerd om de geordende en ongeordende fasen te scheiden. Het systeem paste eerst stateless filtering en mapping parallel toe, en schakelde vervolgens expliciet over naar een sequentiële stream via .sequential() voordat het distinct en sorted aanriep. Deze hybride aanpak beperkte de sequentiële bottleneck tot alleen het stateful eindgedeelte, terwijl 70% van het parallelle doorvoersnelheid werd behouden, terwijl de encounter order werd gegarandeerd.
Het resultaat was een stabiele, deterministische pijplijn die correct de eerste verschijning van elk sensor-evenement identificeerde. De verwerkingssnelheden bleven acceptabel, en het defectpercentage daalde tot nul, terwijl de latentie binnen operationele drempels bleef.
Waarom brengt de forEachOrdered terminaloperatie aanzienlijk hogere overhead met zich mee dan forEach in parallelle streams, en wanneer is het strikt noodzakelijk?
forEach verwerkt elementen zodra ze beschikbaar komen van parallelle threads zonder coördinatie. Deze aanpak maximaliseert de doorvoer maar kan mogelijk uitvoer in thread-arrival-volgorde opleveren. forEachOrdered moet daarentegen de oorspronkelijke encounter order reconstrueren, waardoor het framework resultaten moet bufferen en mogelijk snelle threads moet stilleggen om te wachten op langzamere threads die eerdere elementen bezitten, wat een synchronisatie-bottleneck creëert.
Het is strikt noodzakelijk alleen wanneer de bijeffecten van verwerking de volgorde van de bron moeten observeren. Voorbeelden zijn schrijven naar een positiegevoelige uitvoer zoals een bestand of GUI-lijstmodel. Voor volgorde-ongevoelige bijeffecten zoals loggen of optellen in een gelijktijdige collectie, heeft forEach de voorkeur.
Hoe voorkomt de vereiste voor een associatieve accumulatorfunctie van de reduce operatie subtiele racecondities tijdens parallelle uitvoering, en wat gebeurt er als deze beperking wordt geschonden?
De reduce operatie partitioneert de stream in segmenten, past de accumulator op elk segment in isolatie toe om gedeeltelijke resultaten te produceren, en voegt deze gedeeltelijke resultaten samen met dezelfde accumulator (of een aparte combiner). Associativiteit garandeert dat ((a op b) op c) gelijk is aan (a op (b op c)). Deze eigenschap is vereist omdat de groepering van elementen in segmenten en de volgorde van het samenvoegen van gedeeltelijke resultaten niet-deterministisch en implementatie-afhankelijk is.
Als de operatie niet-associatief is (bijvoorbeeld string concatenatie met een scheidingsteken dat per positie varieert), kan parallelle uitvoering elementen anders groeperen dan sequentiële uitvoering. Dit leidt tot onjuiste resultaten, zoals verwarde scheidingstekens of wiskundig verkeerde sommen voor niet-associatieve aangepaste numerieke types.
Welke specifieke interactie tussen short-circuiting operaties zoals findFirst en oneindige streams zorgt ervoor dat een parallelle stream mogelijk eindeloos kan vastlopen, terwijl een sequentiële stream onmiddellijk zou eindigen?
In een sequentiële stream kan findFirst eindigen zodra de predicaat overeenkomt, zelfs op een oneindige stream. In een parallelle stream splitst het framework de bron in meerdere segmenten die door verschillende threads worden verwerkt. Als het overeenkomende element zich in een segment bevindt dat door een langzame thread wordt verwerkt, moet findFirst wachten tot die thread zijn segment heeft voltooid (of het element heeft gevonden) om te garanderen dat er geen eerder element bestaat in andere segmenten, aangezien het de encounter order moet respecteren.
Als de stream unordered is of findAny in plaats daarvan wordt gebruikt, kan de operatie onmiddellijk eindigen bij elke overeenkomst, waardoor de hoofdthread lopende taken kan annuleren. Kandidaten missen vaak dat findFirst op geordende parallelle oneindige streams in wezen een globale barrière is die kan vastlopen als segmenten vóór de overeenkomst oneindig of computationeel onbeperkt zijn.