L'API Stream fait la distinction entre les opérations sans état (filter, map) et les opérations avec état (sorted, distinct, limit) en fonction de la nécessité de traiter l'entrée entière avant de produire une sortie. Lors de l'exécution en parallèle, le cadre partitionne les données source sur plusieurs threads, chacun traitant un segment de manière indépendante. Si le Spliterator source signale la caractéristique ORDERED, le cadre suppose que l'ordre de rencontre (l'ordre dans lequel les éléments apparaissent dans la source) est significatif et doit être préservé tout au long de la pipeline.
Cependant, les opérations d'état comme distinct reposent sur un état global (un Set d'éléments vus) pour filtrer les doublons. Sans application explicite de l'ordre de rencontre, les threads parallèles peuvent se disputer pour revendiquer les éléments comme étant la "première" occurrence, entraînant une sélection arbitraire de quel doublon survit. De même, sorted nécessite un tri global, mais si le stream est marqué comme non ordonné ou si la source ne possède pas la caractéristique ORDERED, les résultats intermédiaires des threads parallèles peuvent être fusionnés sans préservation de la position. Cela peut donner lieu à différents ordres relatifs d'éléments égaux ou, dans des cas dégénérés, à une apparente non-déterminisme dans la séquence de sortie.
La solution réside dans le respect du contrat Spliterator : si l'ordre de rencontre est important, la source doit déclarer ORDERED, et la pipeline ne doit pas invoquer unordered() avant une opération avec état. Pour distinct, cela garantit que la "première" occurrence dans l'ordre de rencontre est sélectionnée de manière déterministe en traitant les segments de stream dans l'ordre de séquence, réduisant effectivement le parallélisme pour cette étape. Si l'ordre est sans importance, appeler explicitement unordered() permet au cadre d'optimiser en sélectionnant des doublons arbitraires et en fusionnant les résultats partiels sans synchronisation, améliorant ainsi les performances au prix de la déterminisme.
Un système de traitement de télémétrie a ingéré des millions d'événements de capteurs, chacun étiqueté avec un horodatage en nanosecondes et un ID de capteur unique. L'exigence était de dédupliquer les événements par ID de capteur tout en préservant le premier événement chronologique pour chaque ID, puis trier le reste par horodatage. L'implémentation initiale utilisait sensorReadings.parallelStream().distinct().sorted(), supposant que la source ArrayList maintenait l'ordre d'insertion et que distinct préserverait naturellement la première occurrence.
Le problème s'est Manifesté sous forme d'échecs de test intermittents où l'événement "premier" pour un ID de capteur donné serait aléatoirement la deuxième ou la troisième occurrence dans la liste d'origine lors de l'exécution sur du matériel multi-cœur. Après enquête, le problème a été attribué à distinct s'exécutant en parallèle sans enforcement d'ordre de rencontre ; chaque thread traitait un morceau de la liste et conservait sa propre "première" rencontre locale de chaque ID. Lorsque le cadre fusionnait ces résultats partiels, l'ordre global des threads n'était pas garanti, entraînant une sélection arbitraire parmi les premiers locaux de thread.
Trois solutions ont été évaluées. La première approche abandonnait complètement le parallélisme, revenant à un stream séquentiel. Cela a restauré le comportement déterministe, faisant en sorte que le premier événement dans la liste gagne toujours. Cependant, cela a augmentée la latence de traitement de 400 % sous charge de pointe, violant les SLA de débit et nécessitant des mises à niveau matérielles qui n'étaient pas budgétées.
La deuxième approche a inséré .unordered() avant distinct, signalant explicitement qu'un doublon quelconque était acceptable. Cela a maximisé le débit en permettant aux threads de rejeter des doublons arbitraires sans coordination. Malheureusement, cela a violé l'exigence commerciale de préserver la première lecture, rendant l'approche inacceptable pour la traçabilité.
La troisième approche a tiré parti d'un LinkedHashSet comme collecteur en aval via Collectors.toCollection(LinkedHashSet::new) dans une opération collect. Cela a matérialisé le stream dans un ensemble ordonné tout en permettant une décomposition parallèle pour les opérations de filtrage précédentes. Cependant, cela nécessitait d'abandonner l'opération intermédiaire distinct et consommait significativement plus de mémoire pour contenir l'ensemble de travail complet avant la déduplication.
La solution choisie impliquait de restructurer la pipeline pour séparer les phases ordonnées et non ordonnées. Le système a d'abord appliqué un filtrage et un mappage sans état en parallèle, puis a explicitement transitionné à un stream séquentiel via .sequential() avant d'invoquer distinct et sorted. Cette approche hybride a limité le goulot d'étranglement séquentiel à seulement la partie terminale avec état, préservant 70 % du débit parallèle tout en garantissant l'ordre de rencontre.
Le résultat était une pipeline stable et déterministe qui identifiait correctement la première occurrence de chaque événement de capteur. Les vitesses de traitement sont restées acceptables, et le taux de défaut a chuté à zéro tandis que la latence est restée dans les seuils opérationnels.
Pourquoi l'opération terminale forEachOrdered entraîne-t-elle des frais généraux significativement plus élevés que forEach dans les streams parallèles, et quand est-elle strictement nécessaire ?
forEach traite les éléments au fur et à mesure qu'ils deviennent disponibles à partir des threads parallèles sans coordination. Cette approche maximise le débit mais peut potentiellement produire une sortie dans l'ordre d'arrivée des threads. forEachOrdered, en revanche, doit reconstruire l'ordre de rencontre d'origine, ce qui nécessite que le cadre tamponne les résultats et puisse potentiellement ralentir les threads rapides pour attendre ceux plus lents qui détiennent des éléments plus anciens, créant un goulot d'étranglement de synchronisation.
Elle est strictement nécessaire uniquement lorsque les effets secondaires du traitement doivent observer l'ordre source. Les exemples incluent l'écriture dans une sortie sensible à la position comme un fichier ou un modèle de liste GUI. Pour les effets secondaires insensibles à l'ordre comme les journaux ou la somme dans une collection concurrente, forEach est préféré.
Comment l'exigence de l'opération reduce pour une fonction accumulateur associative empêche-t-elle des conditions de course subtiles lors de l'exécution parallèle, et que se passe-t-il si cette contrainte est violée ?
L'opération reduce partitionne le stream en segments, applique l'accumulateur à chaque segment de manière isolée pour produire des résultats partiels, puis combine ces résultats partiels en utilisant le même accumulateur (ou un combineur séparé). L'associativité garantit que ((a op b) op c) est égal à (a op (b op c)). Cette propriété est requise car le regroupement des éléments en segments et l'ordre de combinaison des résultats partiels est non déterministe et dépendant de l'implémentation.
Si l'opération est non associée (par exemple, la concaténation de chaînes avec un délimiteur qui varie selon la position), l'exécution parallèle peut regrouper les éléments différemment de l'exécution séquentielle. Cela entraîne des résultats incorrects tels que des délimiteurs mélangés ou des sommes mathématiquement incorrectes pour des types de nombres personnalisés non associés.
Quelle interaction spécifique entre les opérations à court-circuit comme findFirst et les streams infinis pourrait amener un stream parallèle à potentiellement se bloquer indéfiniment, tandis qu'un stream séquentiel se terminerait immédiatement ?
Dans un stream séquentiel, findFirst peut se terminer dès que le prédicat correspond, même sur un stream infini. Dans un stream parallèle, le cadre divise la source en plusieurs segments traités par différents threads. Si l'élément correspondant se trouve dans un segment traité par un thread lent, findFirst doit attendre que ce thread termine son segment (ou trouve l'élément) pour garantir qu'aucun élément antérieur n'existe dans d'autres segments, car il doit respecter l'ordre de rencontre.
Si le stream est non ordonné ou si findAny est utilisé à la place, l'opération peut se terminer immédiatement dès une correspondance, permettant au thread principal d'annuler les tâches en attente. Les candidats oublient souvent que findFirst sur des streams parallèles infinis ordonnés est effectivement une barrière globale qui peut bloquer si des segments en avance sur la correspondance sont infinis ou computationnellement illimités.