Avant Java 8, la parallélisation du traitement des collections nécessitait une gestion manuelle des Threads ou une soumission explicite à ExecutorService, forçant les développeurs à gérer manuellement la division du travail et la synchronisation. L'introduction de l'API Stream dans Java 8 a abstrait le parallélisme via l'interface Spliterator, qui repose sur des caractéristiques comme SIZED pour indiquer les comptes d'éléments connus. Cette caractéristique permet au framework d'effectuer des coupures binaires équilibrées, créant des arbres de tâches optimaux pour le ForkJoinPool.
Lorsqu'un Spliterator manque de la caractéristique SIZED - commun dans des fonctions génératrices, des streams soutenus par des Iterator, ou des séquences infinies - le framework ne peut pas effectuer de coupures binaires (division par deux) pour créer des arbres de tâches équilibrés. Une coupure aveugle générerait soit des millions de petites tâches (explosion de granularité), causant une surcharge de coordination qui domine le temps d'exécution, soit des morceaux surdimensionnés laissant les threads de travail inactifs pendant qu'un thread traite un énorme arriéré. Cette imprévisibilité brise l’hypothèse fondamentale du parallélisme fork-join : que le travail peut être divisé en sous-tâches à peu près égales.
Le framework utilise le traitement par lots géométriques via l'implémentation par défaut d'IteratorSpliterator. Au lieu de se diviser par moitié, il utilise des tailles de lot augmentant de manière exponentielle (1, 2, 4, 8, jusqu'à MAX_BATCH), ce qui amortit les coûts de division tout en limitant la création de tâches à une profondeur logarithmique. Le ForkJoinPool compense les tailles inconnues en utilisant le vol de travail où les tâches légères sont privilégiées, et l'AbstractTask calcule des signaux d'achèvement sans nécessiter d'informations sur la taille totale. Pour les streams non dimensionnés ordonnés, le pipeline met en mémoire tampon des éléments dans un ArrayList lors de la coupure pour préserver l'ordre de rencontre, échangeant de la mémoire contre la sécurité du parallélisme.
Un système de télémétrie traite des données de capteur en temps réel arrivant via une connexion Socket. Les données arrivent sous forme de flux continu d'objets JSON, et l'exigence commerciale nécessite de parser et de filtrer ces objets en parallèle pour minimiser la latence avant le stockage. Le défi réside dans le taux d'arrivée imprévisible et le volume total de données.
L'implémentation initiale a enveloppé l'InputStream dans un BufferedReader et utilisé lines().parallel(). Cependant, le profilage des performances a révélé que le flux parallèle était considérablement plus lent que le traitement séquentiel en raison d'une surcharge excessive de création de tâches. La cause sous-jacente était le Spliterator de BufferedReader.lines(), qui manque de la caractéristique SIZED et signale initialement Long.MAX_VALUE comme l'estimation, ce qui conduit le framework à créer des micro-tâches pour des lignes individuelles.
Une approche était de mettre en mémoire tampon l'ensemble du flux dans un ArrayList<String> avant le traitement parallèle. Cela fournirait la caractéristique SIZED et permettrait une coupure binaire parfaite entre les cœurs CPU. Cependant, cela introduisait une latence inacceptable — les données ne pouvaient pas être traitées jusqu'à l'arrivée de tout le lot — et créait une pression mémoire sévère lors du traitement de millions d'événements par minute, annulant effectivement le paradigme du streaming.
Une autre solution envisagée était d'implémenter un Spliterator personnalisé qui se divisait toujours en morceaux de taille fixe de exactement 1000 lignes, peu importe le flux sous-jacent. Bien que cela fournisse des tailles de tâche prévisibles, cela échouait lorsque le temps de traitement par ligne variait considérablement; un travailleur pourrait recevoir 1000 objets complexes tandis qu'un autre recevrait 1000 objets simples, entraînant un déséquilibre de charge et des cœurs CPU inactifs attendant la tâche la plus lente.
La solution choisie a impliqué la mise en œuvre d'un Spliterator personnalisé imitant la stratégie de traitement par lots géométriques de la bibliothèque standard. Il traquait une variable batch commençant à 1, doublant à chaque coupure réussie jusqu'à un maximum de 1024, permettant au framework de s'adapter à la longueur réelle du flux sans connaissance préalable. Cette approche a équilibré la surcharge initiale des petites tâches contre l'efficacité des lots plus grands au fur et à mesure de la progression du flux.
L'approche de traitement par lots géométriques a atteint un gain de 3,5 fois sur un système à 8 cœurs par rapport au traitement séquentiel. L'utilisation de la mémoire est restée constante indépendamment de la durée du flux, et la latence est restée faible alors que le traitement commençait immédiatement sans attendre la matérialisation complète. La taille adaptative a prévenu l'explosion de granularité qui avait affecté l'implémentation initiale.
Pourquoi l'encapsulation d'une collection synchronisée dans un flux parallèle réduit souvent les performances par rapport à l'équivalent séquentiel, même pour des opérations intensives en CPU ?
De nombreux candidats supposent que Collections.synchronizedList() ou des implémentations synchronisées de Map sont sûres pour des flux parallèles. Cependant, bien que le Spliterator de ces collections signale SIZED, la synchronisation intrinsèque à chaque accès crée un trafic massif de cohérence de cache. Lorsque plusieurs threads de ForkJoinPool se battent pour le même moniteur pour chaque élément, le coût de la synchronisation et du changement de contexte l'emporte sur tout gain parallèle. La bonne approche nécessite soit d'utiliser ConcurrentHashMap ou CopyOnWriteArrayList (si les écritures sont rares), soit de s'assurer que la collection source est non-interférente et accessible via des caractéristiques Spliterator sans interférence et sûres pour les threads comme CONCURRENT.
Comment la caractéristique ORDERED interagit-elle avec des flux non dimensionnés pour potentiellement sérialiser l'opération terminale, et pourquoi cela s'aggrave-t-il avec sorted() ?
Les candidats manquent souvent que ORDERED combiné à l'absence de SIZED force le framework à mettre en mémoire tampon tous les éléments avant que le traitement puisse être terminé, spécifiquement pour des opérations à état comme sorted() ou distinct(). Sans connaître la taille totale, le framework ne peut pas allouer le tableau final pour toArray() ou les tampons de tri par fusion à l'avance. Il accumule plutôt des éléments dans une liste chaînée ou un ArrayList à redimensionnement dynamique, sérialisant effectivement la phase d'achèvement du pipeline. Cela signifie que l'accélération parallèle est limitée aux étapes de carte/filtre, tandis que l'étape terminale devient un goulet d'étranglement à thread unique attendant l'ensemble du jeu de données.
Quelle violation de contrat spécifique se produit si la méthode trySplit() d'un Spliterator personnalisé retourne un Spliterator qui signale un ensemble de caractéristiques différent de celui du parent ?
Une erreur subtile se produit lorsque les développeurs remplacent trySplit() mais échouent à préserver la cohérence des caractéristiques. Le contrat de Spliterator exige que le spliterator retourné doit avoir les mêmes caractéristiques concernant l'ordre, la distinctivité et la triabilité. Si un parent signale ORDERED mais que l'enfant (résultat de la division) ne le fait pas, les passes d'optimisation du framework Stream peuvent éliminer les étapes de tri ou réorganiser les opérations, conduisant à des résultats incorrects. Les caractéristiques doivent être stables à travers les divisions parce que le pipeline optimise la fusion (par exemple, en combinant filter et map) en fonction de ces indicateurs, et des indicateurs incohérents brisent les relations d’happen-before nécessaires pour la correction parallèle.