22. L'API Stream

Niveau :

Intermédiaire

En programmation fonctionnelle, on décrit le résultat souhaité mais pas comment on obtient le résultat. Ce sont les fonctionnalités sous-jacentes qui se chargent de réaliser les traitements requis en tentant de les exécuter de manière optimisée.

Ce mode de fonctionnement est similaire à SQL : le langage SQL permet d'exprimer une requête mais c'est le moteur de la base de données qui choisir la meilleure manière d'obtenir le résultat décrit. Comme avec SQL, la manière dont on exprime le résultat peut influencer la manière dont le résultat va être obtenu notamment en termes de performance.

Java 8 propose l'API Stream pour mettre en oeuvre une approche de la programmation fonctionnelle sachant que Java est et reste un langage orienté objet.

Le concept de Stream existe déjà depuis longtemps dans l'API I/O, notamment avec les interfaces InputStream et OutputStream. Il ne faut pas confondre l'API Stream de Java 8 avec les classes de type xxxStream de Java I/O. Les streams de Java I/O permettent de lire ou écrire des données dans un flux (sockets, fichiers, ...). Le concept de Stream de Java 8 est différent du concept de flux (stream) de l'API I/O même si l'approche de base est similaire : manipuler un flux d'octets ou de caractères pour l'API I/O et manipuler un flux de données pour l'API Stream. Cette dernière repose sur le concept de flux (stream en anglais) qui est une séquence d'éléments.

L'API Stream facilite l'exécution de traitements sur des données de manière séquentielle ou parallèle. Les Streams permettent de laisser le développeur se concentrer sur les données et les traitements réalisés sur cet ensemble de données sans avoir à se préoccuper de détails techniques de bas niveau comme l'itération sur chacun des éléments ou la possibilité d'exécuter ces traitements de manière parallèle.

L'API Stream de Java 8 propose une approche fonctionnelle dans les développements avec Java. Elle permet de décrire de manière concise et expressive un ensemble d'opérations dont le but est de réaliser des traitements sur les éléments d'une source de données. Cette façon de faire est complètement différente de l'approche itérative utilisée dans les traitements d'un ensemble de données avant Java 8.

Ceci permet au Stream de pouvoir :

• Optimiser les traitements exécutés grâce au lazyness et à l'utilisation d'opérations de type short-circuiting qui permettent d'interrompre les traitements avant la fin si une condition est atteinte
• Exécuter certains traitements en parallèle à la demande du développeur

L'API Stream permet de réaliser des opérations fonctionnelles sur un ensemble d'éléments. De nombreuses opérations de l'API Stream attendent en paramètre une interface fonctionnelle ce qui conduit naturellement à utiliser les expressions lambdas et les références de méthodes dans la définition des Streams. Un Stream permet donc d'exécuter des opérations standards dont les traitements sont exprimés grâce à des expressions lambdas ou des références de méthodes.

Un Stream permet d'exécuter une agrégation d'opérations de manière séquentielle ou en parallèle sur une séquence d'éléments obtenus à partir d'une source dans le but d'obtenir un résultat.

    double tailleMoyenneDesHommes = employes
        .stream()
        .filter(e -> e.getGenre() == Genre.HOMME)
        .mapToInt(e -> e.getTaille())
        .average()
        .orElse(0);

Dans l'exemple ci-dessus, une collection d'éléments de type Employe est utilisée comme source pour un Stream qui va effectuer des opérations de type filter-map-reduce pour calculer la moyenne de la taille des personnes de sexe masculin. La moyenne calculée par la méthode average() est une opération dite de réduction.

Le code utilisé est :

• Concis et clair car il repose sur l'utilisation d'opérations prédéfinies
• Exprimé de manière déclarative (c'est une description du résultat attendu) plutôt que de manière impérative (c'est une description des différentes étapes nécessaires à l'exécution des traitements)

Les éléments traités par le Stream sont fournis par une source qui peut être de différents types :

• Une collection
• Un tableau
• Un flux I/O
• Une chaîne de caractères
• ...

Avec l'API Stream il est possible de déclarer de manière concise des traitements sur une source de données qui seront exécutés de manière séquentielle ou parallèle. Le traitement en parallèle d'un ensemble de données requiert plusieurs opérations :

• La séparation des données en différents paquets plus petits (forking)
• L'exécution dans un thread dédié des traitements sur chaque élément d'un paquet, généralement en utilisant un pool de threads pour limiter la quantité de ressources utilisées
• La combinaison des résultats de chaque paquet pour obtenir le résultat final (joining)

Java 7 propose le framework Fork/Join pour faciliter la mise en oeuvre de ces opérations en parallèle. L'API Stream utilise ce framework pour l'exécution des traitements en parallèle.

L'utilisation de l'API Stream possède donc plusieurs avantages :

• L'exécution, sur un ensemble de données, de traitements définis de manière déclarative
• La quantité de code à produire pour obtenir un traitement similaire reposant sur sa propre itération est réduite
• La possibilité d'exécuter ses traitements en parallèle

Ce chapitre contient plusieurs sections :

• Le besoin de l'API Stream
• Le pipeline d'opérations d'un Stream
• Les opérations intermédiaires
• Les opérations terminales
• Les Collectors
• Les Streams pour des données primitives
• L'utilisation des Streams avec les opérations I/O
• Le traitement des opérations en parallèle
• Les optimisations réalisées par l'API Stream
• Les Streams infinis
• Le débogage d'un Stream
• Les limitations de l'API Stream
• Quelques recommandations sur l'utilisation de l'API Stream

22.1. Le besoin de l'API Stream

Il est fréquent dans les applications de devoir manipuler un ensemble de données. Pour stocker cet ensemble de données, Java propose, depuis sa version 1.1, l'API Collections.

Différentes évolutions ont permis de rendre de plus en plus expressif le code pour traiter ces données. Celles-ci vont être illustrées dans les exemples ci-dessous à l'aide d'un petit algorithme qui va calculer la somme des valeurs inférieures à 10 dans une collection contenant les 12 premiers entiers.

Avec Java 5, il est possible d'utiliser un Iterator avec les generics.

    List<Integer> entiers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12);
    Iterator<Integer> it = entiers.iterator();
    long somme = 0;
    while (it.hasNext()) {
      int valeur = it.next();
      if (valeur < 10) {
        somme += valeur;
      }
    }
    System.out.println(somme);

Java 5 a aussi introduit une nouvelle syntaxe de l'instruction for qui permet de faciliter l'écriture du code requis pour parcourir une collection

   long somme = 0;
   for (int valeur : entiers) {
     if (valeur < 10) {
       somme += valeur;
     }
   }
   System.out.println(somme);

Dans les exemples précédents, le parcours des éléments de la collection est fait à la main sous la forme d'une itération externe (external iteration) qui contient la logique de traitements. Finalement le code à produire pour réaliser cette tâche relativement simple est assez important.

Java 8 propose la méthode forEach() dans l'interface Collection qui attend en paramètre une interface fonctionnelle de type Consumer.

    LongAdder somme = new LongAdder();
    entiers.forEach(valeur -> {
      if (valeur < 10) {
        somme.add(valeur);
      }
    });
    System.out.println(somme);

Dans les exemples précédents, le code de l'algorithme est exécuté séquentiellement. Il serait beaucoup plus complexe et verbeux pour arriver à le faire exécuter en parallèle (en imaginant que le volume de données à traiter soit beaucoup plus important).

L'ajout du support de traitement en parallèle dans l'API Collections aurait été très compliqué : le choix a été fait de définir une nouvelle API permettant de :

• Faciliter l'exécution de traitements sur un ensemble de données
• Réduire la quantité de code nécessaire pour le faire
• Permettre d'exécuter des opérations en parallèle de manière très simple

Pour offrir une solution à cette problématique et proposer la possibilité de simplifier la réalisation d'opérations plus ou moins complexes sur des données, Java 8 propose l'API Stream.

    long somme = entiers.stream()
                        .filter(v -> v <10)
                        .mapToInt(i -> i)
                        .sum();
    System.out.println(somme);

Elle permet notamment très facilement d'exécuter ces traitements en parallèle.

    long somme = entiers.parallelStream()
                        .filter(v -> v < 10)
                        .mapToInt(i -> i)
                        .sum();
    System.out.println(somme);

L'approche proposée par Java 8 est de laisser l'API réaliser l'itération (internal iteration).

L'utilisation de l'API Stream permet de se concentrer sur l'essentiel (la logique des traitements) et de l'exprimer sous une forme expressive et succincte. L'exemple de traitement tient maintenant sur une ligne de code.

Ainsi avant Java 8, la seule manière de traiter les éléments d'une collection est d'itérer sur ceux-ci généralement en utilisant un Iterator explicitement en utilisant une boucle while ou implicitement en utilisant une boucle de type for each. Cette approche est celle de la programmation impérative avec laquelle il est nécessaire de coder toutes les étapes de l'algorithme à utiliser. Généralement cela implique une écriture verbeuse du code de ces traitements puisque celui-ci détaille ce que doivent faire les fonctionnalités. Dans la plupart des cas, il serait moins verbeux de décrire les résultats attendus et de laisser l'application exécuter à sa manière les fonctionnalités correspondantes.

package fr.jmdoudoux.dej.streams;

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

public class TestBoucle {

  public static void main(String[] args) {
    List<Personne> personnes = new ArrayList<>(6);
    personnes.add(new Personne("p1", Genre.HOMME, 176));
    personnes.add(new Personne("p2", Genre.HOMME, 190));
    personnes.add(new Personne("p3", Genre.FEMME, 172));
    personnes.add(new Personne("p4", Genre.FEMME, 162));
    personnes.add(new Personne("p5", Genre.HOMME, 176));
    personnes.add(new Personne("p6", Genre.FEMME, 168));

    long total = 0;
    int nbPers = 0;
    for (Personne personne : personnes) {
      if (personne.getGenre() == Genre.FEMME) {
        nbPers++;
        total += personne.getTaille();
      }
    }
    double resultat = (double) total / nbPers;
    System.out.println("Taille moyenne des femmes = " + resultat);
  }
}

Les traitements de l'exemple précédent peuvent être réécrits en utilisant l'API Stream.

package fr.jmdoudoux.dej.streams;

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

public class TestBoucle {

  public static void main(String[] args) {
    List<Personne> personnes = new ArrayList<>(6);
    personnes.add(new Personne("p1", Genre.HOMME, 176));
    personnes.add(new Personne("p2", Genre.HOMME, 190));
    personnes.add(new Personne("p3", Genre.FEMME, 172));
    personnes.add(new Personne("p4", Genre.FEMME, 162));
    personnes.add(new Personne("p5", Genre.HOMME, 176));
    personnes.add(new Personne("p6", Genre.FEMME, 168));
    
    resultat = personnes
      .stream()
      .filter(p -> p.getGenre() == Genre.FEMME)
      .mapToInt(p -> p.getTaille())
      .average()
      .getAsDouble();
    System.out.println("Taille moyenne des femmes = " + resultat);
  }
}

Dans l'exemple ci-dessus, les données traitées par le Stream subissent plusieurs opérations :

• Un filtre qui utilise le Predicat passé en paramètre sous la forme d'une expression Lambda pour ne conserver que les données concernant le genre féminin
• Le nouveau Stream obtenu est transformé en utilisant la méthode mapToInt() qui utilise l'interface fonctionnelle ToIntFunction passée en paramètre sous la forme d'une expression lambda pour obtenir un nouveau Stream qui ne contient que les tailles des personnes féminines sous la forme d'entiers
• La méthode d'agrégation average() du Stream de type IntStream obtenu permet de calculer la moyenne de ces valeurs entières
• Le résultat est une instance de type OptionalDouble dont la méthode getAsDouble() permet d'obtenir la valeur sous la forme d'une valeur flottante de type double si une valeur est présente

L'API Stream propose donc d'ajouter une manière plus expressive et une approche fonctionnelle au langage Java pour le traitement de données :

• Il n'y a plus de code pour itérer sur chacun des éléments
• Des opérations sont utilisées en leur passant en paramètre des expressions lambdas ou des références de méthode pour indiquer le détail de la tâche à accomplir

L'interface Stream<T> propose une liste définie d'opérations qu'il sera possible de réaliser. La plupart de ces opérations renvoient une instance de type Stream<T> pour permettre de combiner ces opérations.

22.1.1. L'API Stream

L'API Stream est contenue dans le package java.util.stream. Ce package est composée de 10 interfaces, deux classes et une énumération.

Les interfaces de l'API sont :

L'interface BaseStream<T,S extends BaseStream<T,S>> est l'interface mère des principales interfaces utilisées :

• Stream<T> : interface permettant le traitement d'objets de type T de manière séquentielle ou parallèle
• IntStream : spécialisation pour traiter des données de type int
• LongStream : spécialisation pour traiter des données de type long
• DoubleStream : spécialisation pour traiter des données de type double

Si les éléments du Stream sont des données numériques alors il n'est pas recommandé d'utiliser un Stream<T> où T est de type Double, Long ou Integer. Les opérations sur ces objets vont nécessiter de l'autoboxing. Dans ces cas, il est préférable d'utiliser un DoubleStream, IntStream ou LongStream pour améliorer les performances des traitements de ces données et profiter de certaines opérations de réductions dédiés aux type primitifs tels que la somme ou la moyenne.

L'API définit plusieurs classes :

22.1.2. Le rôle d'un Stream

Il est très fréquent de vouloir traiter des données stockées dans une collection. Cependant, l'exécution de traitements sur une collection présente plusieurs inconvénients :

• Elle requière de nombreuses lignes de code notamment quelques-unes pour gérer l'itération sur les différents éléments ou pour réaliser des opérations de base comme la somme de valeurs ou la recherche du plus petit ou plus grand élément
• Il n'est pas facile de faire exécuter ces traitements de manière parallèle : ceci pourrait cependant être particulièrement utile lors du traitement d'une très large quantité de données. Et les quantités de données à traiter augmente de manière croissante globalement dans le temps

    List<Employe> employeMasculins = new ArrayList<>();
    for (Employe e : employes) {
      if (e.getGenre() == Genre.HOMME) {
        employeMasculins.add(e);
      }
    }
    
    Collections.sort(employeMasculins, new Comparator<Employe>() {
      @Override
      public int compare(Employe p1, Employe p2) {
        return p2.getTaille() - p1.getTaille();
      }
    });

    List<String> nomEmployes = new ArrayList<>();
    for (Personne p : employeMasculins) {
      nomEmployes.add(p.getNom());
    }
    for (String nom : nomEmployes) {
      System.out.println(nom);
   }

Un Stream peut être vu comme une abstraction qui permet de réaliser des opérations définies sur un ensemble de données.

Le but principal d'un Stream est de pouvoir décrire de manière expressive des traitements à exécuter sur un ensemble de données en limitant au maximum le code à produire pour y arriver. Globalement cela passe par une description de ce que l'on veut faire mais pas comment cela va se faire.

De la même manière que SQL permet de définir les données à obtenir et la façon dont on souhaite qu'elles soient restituées, les Streams permettent de décrire des opérations à réaliser sur un ensemble de données.

La déclaration d'un Stream se fait en plusieurs étapes :

• Création d'un Stream à partir d'une source : collection ou autre
• Définition de zéro, une ou plusieurs opérations intermédiaires qui renvoient toutes un Stream
• Définition d'une opération terminale qui va permettre d'obtenir le résultat des traitements et qui va clôturer le Stream

    List<String> nomEmployes = 
      employes.stream()
              .filter(e -> e.getGenre() == Genre.HOMME)
              .sorted(Comparator.comparingInt(Employe::getTaille)
                                .reversed())
              .map(Personne::getNom)
              .collect(Collectors.toList());

    for (String nom : nomEmployes) {
      System.out.println(nom);
    }

Le code Java 8 est plus compact car il ne contient que les informations indispensables à la bonne exécution des traitements :

• On obtient un Stream à partir de la collection contenant les données
• On applique plusieurs opérations filtre pour ne conserver sur les employés masculins, trie ces employés par taille décroissante et extraction des noms
• On les insère dans une collection de type List

De plus, comme la manière d'exécuter ces traitements est laissée à l'implémentation du Stream, il est possible que celui-ci les exécute en parallèle sur simple demande en remplaçant la méthode stream() par parallelStream().

    List<String> nomEmployes = 
      employes.parallelStream()
              .filter(e -> e.getGenre() == Genre.HOMME)
              .sorted(Comparator.comparingInt(Employe::getTaille)
                                .reversed())
              .map(Personne::getNom)
              .collect(Collectors.toList());

    for (String nom : nomEmployes) {
      System.out.println(nom);
    }

Un Stream parallèle va en interne paralléliser l'exécution du pipeline d'opérations en répartissant les données sur plusieurs threads pour utiliser les coeurs présents sur la machine.

22.1.3. Les concepts mis en oeuvre par les Streams

Une définition simple d'un Stream pourrait être : « l'application d'un ensemble d'opérations sur une séquence d'éléments issus d'une source dans le but d'obtenir un résultat ».

Cette définition met en avant plusieurs concepts :

• Un ensemble d'opérations : ce sont les traitements ordonnés qui seront exécutés. Chaque opération permet d'effectuer une tâche. Les opérations communes en programmation fonctionnelle sont proposées : filtre (filter), transformation (map), réduction (reduce), recherche (find), correspondance (match), tri (sorted), ...
• Pipeline d'opérations : de nombreuses opérations renvoient un Stream ce qui permet de les enchaîner et éventuellement de réaliser des optimisations
• Une séquence d'éléments : ce sont les données d'un certain type sur lesquelles les opérations vont être appliquées mais le Stream ne stocke pas ses données car elles sont obtenues à la demande
• Une source : elle permet de fournir les données à traiter au Stream comme une collection, un tableau ou tout autre ressource capable de fournir un Stream

De plus, un Stream utilise d'autres concepts lors de sa mise en oeuvre :

• Lazyness : les données sont consommées de manière tardive
• Short-circuiting : certaines opérations peuvent interrompre le traitement des éléments
• Parallélisation possible des traitements
• Internal iteration : c'est l'API qui réalise l'itération sur les données à traiter

22.1.3.1. Le mode de fonctionnement d'un Stream

Un Stream permet d'exécuter des opérations sur un ensemble de données obtenues à partir d'une source afin de générer un résultat.

package fr.jmdoudoux.dej.streams;

import java.util.Arrays;
import java.util.List;

public class JavaApplication1 {

  public static void main(String[] args) {
    List<String> maListe = Arrays.asList("a1", "a2", "b2", "b1", "c1");
    maListe.stream()
           .filter(s -> s.startsWith("b"))
           .map(String::toUpperCase)
           .sorted()
           .forEach(System.out::println);
  }
}

B1
B2

Il existe deux types d'opérations : les opérations intermédiaires et les opérations terminales. L'ensemble des opérations effectuées par un Stream est appelé pipeline d'opérations (operation pipeline).

La plupart des opérations d'un Stream attendent en paramètre une interface fonctionnelle pour définir leur comportement. Ces paramètres peuvent ainsi être exprimés en utilisant une expression lambda ou une référence de méthode.

Une opération peut être avec ou sans état.

Il est important que :

• Les opérations exécutées par le Stream ne modifient pas la source de données utilisée par le Stream
• Les données de la source de données ne doivent pas être modifiées durant leur traitement par le Stream car cela pourrait avoir un impact sur leur exécution

22.1.3.2. Les opérations pour définir les traitements d'un Stream

Les opérations d'un Stream permettent de décrire des traitements, potentiellement complexes, à exécuter sur un ensemble de données. Elles sont définies dans l'interface java.util.stream.Stream grâce à de nombreuses méthodes.

Il existe deux catégories d'opérations :

• Opérations intermédiaires : elles peuvent être enchaînées car elles renvoient un Stream
• Opérations terminales : elles renvoient une valeur différente d'un Stream (ou pas de valeur) et ferme le Stream à la fin de leur exécution

Les opérations intermédiaires ne réalisent aucun traitement tant que l'opération terminale n'est invoquée. Elles sont dites lazy ce qui permettra éventuellement d'effectuer des optimisations dans leur exécution comme car exemple de réaliser celle-ci dans une même itération.

package fr.jmdoudoux.dej.streams;

import java.util.Arrays;
import java.util.List;
import static java.util.stream.Collectors.toList;

public class TestStream2 {

  public static void main(String[] args) {
    List<Integer> nombres = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8);
      
    List<Integer> troisPremierNombrePairAuCarre = 
      nombres.stream()
             .filter(n -> {
                 System.out.println("filter " + n);
                 return n % 2 == 0;
              })
             .map(n -> {
                 System.out.println("map " + n);
                 return n * n;
              })
             .limit(3)
             .collect(toList());

    System.out.println("");
    troisPremierNombrePairAuCarre.forEach(System.out::println);
  }
}

filter 1
filter 2
map 2
filter 3
filter 4
map 4
filter 5
filter 6
map 6
4
16
36

Dans l'exemple ci-dessus, plusieurs optimisations sont mises en oeuvre par le Stream. Tous les éléments ne sont pas parcourus. Les opérations ne sont invoquées que le nombre de fois requis sur des éléments pour obtenir le résultat :

• L'opération de filtre n'est invoquée que 6 fois
• L'opération de transformation n'est invoquée que 3 fois

Ceci est possible car l'opération limit(3) ne renvoie qu'un Stream qui contient au maximum le nombre d'éléments précisé en paramètre.

Les opérations sont regroupées en une seule itération qui traite chaque élément en lui appliquant le pipeline d'opérations : les opérations n'itèrent pas individuellement sur les éléments du Stream.

Ceci permet d'optimiser au mieux les traitements à réaliser pour obtenir le résultat de manière efficiente.

L'utilisation d'un Stream implique généralement trois choses :

• Une source qui va alimenter le Stream avec les éléments à traiter
• Un ensemble d'opérations intermédiaires qui vont décrire les traitements à effectuer
• Une opération terminale qui va exécuter les opérations et produire le résultat

Les opérations proposées par un Stream sont des opérations communes :

• Pour filtrer des données
• Pour rechercher une correspondance avec des éléments
• Pour transformer des éléments
• Pour réduire les éléments et produire un résultat

Pour filtrer des données, un Stream propose plusieurs opérations :

• filter(Predicate) : renvoie un Stream qui contient les éléments pour lesquels l'évaluation du Predicate passé en paramètre vaut true
• distinct() : renvoie un Stream qui ne contient que les éléments uniques (elle retire les doublons). La comparaison se fait grâce à l'implémentation de la méthode equals()
• limit(n) : renvoie un Stream que ne contient comme éléments que le nombre fourni en paramètre
• skip(n) : renvoie un Stream dont les n premiers éléments sont ignorés

Pour rechercher une correspondance avec des éléments, un Stream propose plusieurs opérations :

• anyMatch(Predicate) : renvoie un booléen qui précise si l'évaluation du Predicate sur au moins un élément vaut true
• allMatch(Predicate) : renvoie un booléen qui précise si l'évaluation du Predicate sur tous les éléments vaut true
• noneMatch(Predicate) : renvoie un booléen qui précise si l'évaluation du Predicate sur tous les éléments vaut false
• findAny() : renvoie un objet de type Optional qui encapsule un élément du Stream s'il existe
• findFirst() : renvoie un objet de type Optional qui encapsule le premier élément du Stream s'il existe

Pour transformer des données, un Stream propose plusieurs opérations :

• map(Function) : applique la Function fournie en paramètre pour transformer l'élément en créant un nouveau
• flatMap(Function) : applique la Function fournie en paramètre pour transformer l'élément en créant zéro, un ou plusieurs éléments

Pour réduire les données et produire un résultat, un Stream propose plusieurs opérations :

• reduce() : applique une Function pour combiner les éléments afin de produire le résultat
• collect() : permet de transformer un Stream qui contiendra le résultat des traitements de réduction dans un conteneur mutable

22.1.4. La différence entre une collection et un Stream

Il est tentant de voir un Stream comme une collection. Les notions de collections et de Streams concernent toutes les deux une séquence d'éléments. De manière grossière, une collection permet de stocker cette séquence d'éléments et un Stream permet d'exécuter des opérations sur cette séquence d'éléments.

Bien que les Collections et les Streams semblent avoir des similitudes, ils ont des objectifs différents :

• Les collections permettent de gérer et de récupérer des données qu'elles stockent
• Les Streams assurent l'exécution lazy de traitements déclarés sur des données d'une source

Les Streams diffèrent donc de plusieurs manières des Collections :

• Un Stream n'est pas une structure qui stocke des données. Elle permet de traiter les différents éléments d'une source en appliquant différentes opérations
• Un Stream permet de mettre en oeuvre la programmation fonctionnelle : le résultat d'une opération ne doit pas modifier l'élément sur lequel elle opère ni aucun autre élément de la source. Par exemple, une opération de filtre ne doit pas retirer des éléments de la source mais créer un nouveau Stream qui contient uniquement les éléments filtrés
• De nombreuses opérations d'un Stream attendent en paramètre une expression lambda
• Il n'est pas possible d'accéder à un élément grâce à un index : seul le premier peut être obtenu ou il faut exporter le résultat sous la forme d'une collection ou d'un tableau
• Il est possible de paralléliser l'exécution des opérations. Cette exécution parallèle utilise alors le framework Fork/Join
• Un Stream n'a pas forcement de taille fixe : le nombre d'éléments à traiter peut potentiellement être infini. Ils peuvent consommer des données jusqu'à ce qu'une condition soit satisfaite : des méthodes comme limit() ou findFirst() peuvent alors permettre de définir une condition d'arrêt
• Une opération ne peut consommer qu'une seule fois un élément. Un nouveau Stream doit être recréer pour traiter de nouveau un élément comme le fait un Iterator
• Les Streams ne stockent pas de données. Ils transportent et utilisent les données issues d'une source qui les stockent ou les génèrent
• Les Streams exécutent un pipeline d'opérations sur les données de leur source
• Les Streams sont de nature fonctionnel : les opérations d'un Stream produisent des résultats mais ne devrait pas modifier les données de leur source
• Les opérations intermédiaires des Streams sont exécutées de manière lazy. Ce mode de fonctionnement permet d'effectuer des optimisations lors de l'exécution en un seul passage du pipeline d'opérations

Une collection est une structure qui stocke un ensemble de données en mémoire. Les traitements sur une collection, sans utiliser les Streams, nécessitent de réaliser une itération sur tout ou partie des éléments de la collection par exemple en utilisant l'instruction for.

    List<Personne> personnes = // ... code pour obtenir une liste de personnes

    List<Long> ids = new ArrayList<>(personnes.size());
    for(Personne p: personnes){
      ids.add(p.getId());
    }

Avec un Stream, les opérations à réaliser sont décrites mais l'itération ou les itérations pour exécuter ces traitements sont réalisées en interne lors de l'exécution de la méthode terminale du Stream.

    List<Personne> personnes = // ... code pour obtenir une liste de personnes

    List<Long> ids = personne.stream()
                             .map(Personne::getId)
                             .collect(toList());

Le code ci-dessus décrit simplement les opérations à réaliser par le Stream. La façon dont ces traitements seront exécutés est à la charge de l'implémentation du Stream.

Dans les deux exemples, le résultat est le même.

Les collections sont des structures en mémoire qui permettent de stocker des données : toutes les valeurs doivent être ajoutées à la collection avant de pouvoir les traiter. Un Stream ne stocke pas de données : il va obtenir les données à traiter à la demande à partir d'une source. Il effectue des traitements sur les données de cette source sur la base de l'exécution d'un pipeline d'opérations.

Le fait que l'API Stream réalise elle-même en interne l'itération sur les éléments de la source de données permet que la plupart des opérations soient exécutés de manière lazy. Ceci permet à l'API de réaliser éventuellement des optimisations.

Un Stream est un consommable : il consomme les données de la source une fois que son opération terminale est invoquée.

Un Stream peut assurer un traitement séquentiel ou en parallèle de ces traitements : le traitement en parallèle peut être très intéressant d'un point de vue performance notamment dans le cas du traitement d'un gros volume de données.

22.1.5. L'obtention d'un Stream

La création d'un Stream se fait nécessairement à partir d'une source de données. Cette source va permettre de fournir les différents éléments à la demande du Stream pour pouvoir les traiter.

Les Streams peuvent utiliser des données issues d'une source finie ou infinie d'éléments.

Il existe différentes façons de construire un Stream à partir de différentes sources :

• Collection
• Tableau
• Ensemble de données
• Fichier
• ...

Il est aussi possible d'utiliser une Function pour produire un nombre infini d'éléments comme source de données.

Chacune des classes pouvant servir de source propose une ou plusieurs méthodes pour créer un Stream :

Il est également possible que des sources soient proposées par des bibliothèques tierces ou soient développées par ses propres soins.

22.1.5.1. La création d'un Stream à partir de ses fabriques

L'interface Stream propose plusieurs fabriques pour créer un Stream.

La méthode of() de l'interface Stream attend en paramètre un varargs d'objets.

    Stream<Integer> stream = Stream.of(new Integer[] { 1, 2, 3});

Attention, ce sont des objets qui sont attendus. Il n'est donc pas possible de fournir en paramètre un tableau de type entier primitif.

    Stream<Integer> stream = Stream.of(new int[] { 1, 2, 3});

C:\java\workspace\TestStreams\src>javac com/jmdoudoux/dej/streams/TestStream.java
com\jmdoudoux\dej\streams\TestStream.java:9:
error: incompatible types: inference variable T has incompatible bounds
    Stream<Integer> stream1 = Stream.of(new int[] { 1, 2, 3 });
                                       ^
    equality constraints: Integer
    lower bounds: int[]
  where T is a type-variable:
    T extends Object declared in method <T>of(T)
1 error

Il est cependant possible de passer en paramètre de la méthode of() de l'interface IntStream un tableau d'entiers primitifs.

    IntStream stream = IntStream.of(new int[] { 1, 2, 3});

Les méthodes iterate() et generate() de l'interface Stream attendent en paramètre une Function. Les éléments générés sont calculés à la demande en invoquant la fonction pour générer la prochaine valeur. Sans contrainte particulière, cette génération se poursuit à l'infini. Dans ce cas, on parle de Stream infini : c'est un Stream dont la source ne connait pas le nombre d'éléments qui sera traité à l'avance.

Par exemple, la méthode iterate() attend en paramètre une valeur initiale et un UnaryOperator qui sera invoqué pour générer chaque nouvelle valeur.

    Stream<Integer> entiers = Stream.iterate(0, n -> n + 1);
    entiers.forEach(System.out::println);

Ce traitement va incrémenter une valeur entière et l'afficher sur la console.

Il est bien sûr possible de définir une condition d'arrêt des traitements du Stream basée sur le nombre d'éléments traités en utilisant la méthode limit() avec en paramètre le nombre souhaité.

    Stream<Integer> entiers = Stream.iterate(0, n -> n + 1);
    entiers.limit(10)
           .forEach(System.out::println);

22.1.5.2. L'obtention d'un Stream à partir d'une collection

Les traitements sur les données d'une collection doivent être réalisés en itérant explicitement sur chacun des éléments. Avec les Streams l'itération est effectué par les traitements de l'API.

L'interface Collection propose deux méthodes pour obtenir un Stream dont la source sera la collection :

Un Stream peut donc être obtenu à partir de n'importe quelle implémentation de l'interface Collection en utilisant la méthode stream().

List<String> elements = new ArrayList<String>();
elements.add("element1");
elements.add("element2");
elements.add("element3");
Stream<String> stream = elements.stream();

22.2. Le pipeline d'opérations d'un Stream

Un Stream exécute une combinaison d'opérations pour former un pipeline.

Les opérations sont des méthodes définies dans l'interface Stream :

• Les méthodes intermédiaires : map(), mapToInt(), flatMap(), mapToDouble(), filter(), distinct(), sorted(), peek(), limit(), skip(), parallel(), sequential(), unordered()
• Les méthodes terminales : forEach(), forEachOrdered(), toArray(), reduce(), collect(), min(), max(), count(), anyMatch(), allMatch(), noneMatch(), findFirst(), findAny(), iterator()

La plupart des opérations d'un Stream attendent en paramètres une interface fonctionnelle qui sont généralement définies dans le package java.util.function. Cela permet d'utiliser une expression Lambda pour décrire les traitements qui seront réalisés par l'opération.

Le traitement de données par un Stream se fait en deux étapes :

• Configuration en invoquant des méthodes intermédiaires
• Exécution des traitements en invoquant la méthode terminale

Remarque : aucun traitement n'est effectué lors de l'invocation des méthodes intermédiaires. Ces méthodes sont dites lazy.

Dès qu'une méthode terminale est invoquée, le Stream est considéré comme consommé et plus aucune de ces méthodes ne peut être invoquée. Une méthode terminale peut produire une ou plusieurs valeurs ou opérer des effets de bord (forEach).

L'exemple ci-dessous créé une liste des 5 premières personnes dont le nom commence par un 'A'.

List<String> prenomsCommencantParA = personnes
    .stream()
    .map(Personne::getNom)
    .filter(nom->nom.startsWith("a"))
    .limit(5)
    .collect(Collectors.toList());

Cet exemple utilise trois opérations intermédiaires :

• map() : transforme chaque élément de type Personne pour renvoyer uniquement le nom
• filter() : permet de limiter les éléments pour ne conserver que ceux qui commencent par 'A'
• limit() : arrête les traitements dès que le Stream contient 5 éléments

L'opération terminale collect() transforme les résultats de l'exécution dans une collection de type List.

Les opérations d'un Stream sont réparties en deux catégories :

• Les opérations intermédiaires (intermediate operation) : zéro, une ou plusieurs opérations intermédiaires par Stream. Elles produisent un Stream et sont toujours évaluées de manière lazy
• Les opérations terminales (terminal operation) : une seule opération terminale par Stream. Elles produisent une valeur ou des effets de bord

Une opération intermédiaire produit et renvoie un objet de type Stream. Ces opérations peuvent donc être chaînées pour définir le pipeline d'opérations.

Le traitement de ces méthodes n'est pas exécuté tant qu'une méthode terminale n'est pas invoquée. L'invocation d'une méthode intermédiaire permet simplement d'ajouter l'opération au pipeline : seule l'invocation de l'opération terminale va lancer la récupération des données et leur utilisation au travers du pipeline d'opérations.

Chaque opération intermédiaire :

• Renvoie un nouveau Stream : elle produit des éléments qui seront utilisés par l'opération suivante mais ne peut jamais produire le résultat final d'un Stream
• Opère de manière lazy : renvoie un nouveau Stream qui contiendra les éléments résultant de l'exécution de l'opération. Le parcours des éléments de la source ne commence pas avant l'invocation de la méthode terminale

La plupart de ces opérations intermédiaires d'un Stream fonctionnent en mode lazy : il diffère le plus tardivement ses traitements jusqu'à ce que les résultats soient nécessaires. Le mode lazy permet des optimisations, par exemple : il n'est pas nécessaire de parcourir tous les éléments pour trouver le premier qui correspondent à un critère donné. Ce type d'optimisation est particulièrement utile notamment si le nombre d'éléments à traiter est important

Il est important de comprendre que les opérations intermédiaires s'exécutent en mode lazy. Leur invocation ne provoque aucun traitement : c'est l'invocation de la méthode terminale qui provoque le traitement des données par les opérations du Stream.

Ce mode de fonctionnement possède plusieurs avantages :

• Généralement permet l'invocation des opérations en une passe
• Ceci est particulièrement utile lorsque la quantité de données à traiter par le Stream est importante
• L'API Stream peut optimiser les opérations exécutées notamment en utilisant des opérations de type short-circuiting. Ce type d'opération peut interrompre les traitements du Stream lorsqu'une condition est satisfaite

Pour illustrer ce comportement, l'exemple ci-dessous utilise un Stream sur une collection de prénoms pour les mettre en majuscule, ne conserver que ceux qui commencent par un 'A' et ne prendre que les deux premiers pour les afficher.

  prenoms.stream()
         .map(String::toUpperCase)
         .filter(p -> p.startsWith("A"))
         .limit(2)
         .forEach(System.out::println);

ANDRE
ALBERT

Bien que l'opération limit() soit la dernière opération intermédiaire, elle a une influence sur l'exécution. Une fois que la condition est atteinte (deux prénoms commençant par A) alors les traitements effectués par le Stream sont interrompus. Ceci est possible car le pipeline d'opérations est appliqué sur chaque donnée.

  public static void main(String[] args) {
    List<String> prenoms = Arrays.asList("andre", "benoit", "albert", "thierry", "alain",
      "jean");
    prenoms.stream()
           .map(p -> {
              afficher("map    ", p);
              return p.toUpperCase();})
           .filter(p -> {
                afficher("filter ", p);
                 return p.startsWith("A");})
           .peek(p -> {
              afficher("limit ", p);})
           .limit(2)
           .forEach(System.out::println);
  }

  public static void afficher(String message, String p) {
    System.out.println(message + " : " + p);
  }

map    : andre
filter : ANDRE
limit  : ANDRE
ANDRE
map    : benoit
filter : BENOIT
map    : albert
filter : ALBERT
limit  : ALBERT
ALBERT

Cette façon d'exécuter les traitements permet d'améliorer les performances notamment lorsque la quantité de donnée est importante. Au lieu de réaliser la transformation et le filtre sur toutes les données pour ne conserver que les trois premières, elles sont appliquées sur les données jusqu'à ce que la condition soit respectée.

Un pipeline d'opérations doit impérativement se terminer par une opération terminale.

Les opérations terminales lancent les traitements du Stream pour produire le résultat de l'exécution du pipeline d'opérations ou des effets de bords sur les données consommées :

• Renvoyer le résultat, généralement sous la forme d'une instance de type Optional<T> (elles ne renvoient jamais un Stream)
• Créer une collection qui contient les résultats
• Exécuter des traitements sur les résultats (méthode forEach())

Une fois qu'une méthode terminale est invoquée, elle lance les traitements et le Stream ne pourra alors plus être utilisé. Une fois que l'opération terminale est exécutée, le Stream est considéré comme consommé et il ne peut plus être utilisé. Si les données de la source doivent de nouveau être parcourues, il faut obligatoirement recréer un nouveau Stream à partir de la source.

Les opérations terminales ont pour rôle de terminer le pipeline d'opérations :

• Elles lancent l'exécution des traitements
• Elles produisent le résultat
• Et elles ferment le Stream

Certaines méthodes intermédiaires (limit(), skip(), ...) ou terminales (anyMatch(), allMatch(), noneMatch(), findFirst(), findAny(), ...) sont de type « short-circuiting » :

• Pour des opérations intermédiaires ce sont des opérations qui produisent un Stream dont le nombre d'éléments est fini à partir d'un Stream dont le nombre d'éléments est potentiellement infini
• Pour des opérations terminales ce sont des opérations qui peuvent se terminer dans un temps fini à partir d'un Stream dont le nombre d'éléments est potentiellement infini

Si le pipeline contient une opération de type short-circuiting, les méthodes intermédiaires suivantes ne seront exécutées que lorsque la méthode short circuit pourra être évaluée.

Avoir une opération de type short-circuit dans le pipeline du Stream est une condition nécessaire mais pas toujours suffisante pour permettre de terminer l'exécution d'un Stream dont le nombre d'éléments est infini.

22.3. Les opérations intermédiaires

Une opération intermédiaire renvoit toujours un Stream.

Le pipeline d'opérations d'un Stream peut avoir zéro, une ou plusieurs opérations intermédiaires.

Les opérations intermédiaires peuvent être réparties en deux groupes :

• Les opérations stateless ne conserve pas d'état lors du traitement d'un élément par rapport au précédent. Chaque élément est donc traité indépendamment des autres.
• Les opérations stateful conserve un état par rapport aux éléments traités précédemment lors du traitement d'un élément. Certaines opérations stateful doivent traiter l'ensemble des éléments avant de pouvoir créer leur résultat. Ceci peut avoir des conséquences lors de l'utilisation d'opérations stateful dans un Stream avec traitements en parallèle : cela peut nécessiter plusieurs itérations sur les données.

L'API Stream définit plusieurs opérations intermédiaires :

Les opérations sans état (stateless) comme map() ou filter() renvoient simplement le résultat de l'exécution sur l'élément courant dans le Stream pour être traité par l'opération suivante.

Les opérations avec état (stateful) peuvent avoir besoin de conserver des informations relatives au traitement des éléments précédent voir de tous les éléments pour pouvoir produire le Stream contenant les éléments à traiter par l'opération suivante.

22.3.1. Les méthodes map(), mapToInt(), mapToLong et mapToDouble()

Une opération de mapping permet de réaliser une transformation des éléments traités par le Stream.

Parfois, il est nécessaire de transformer les éléments d'un Stream. La méthode map() permet d'opérer une transformation en passant chacun des éléments du Stream à la Function fournie en paramètre. Le résultat est la production d'un nouveau Stream contenant le résultat de l'application de la fonction à chaque élément. C'est une transformation de type un pour un.

L'interface Stream<T> propose plusieurs opérations pour faire des transformations de type map.

La méthode map(Function< ? super T, ? super R>) attend en paramètre une Function ayant en argument un objet de type T et renvoie un objet de type R. Cette Function va être appliquée sur tous les éléments du Stream pour créer pour chacun un nouvel élément qui sera inclus dans le Stream en résultat. L'implémentation de l'interface fonctionnelle fournie en paramètre doit être :

• Sans état : aucune information relative aux éléments traités précédemment ne doit être conservée
• Ne pas modifier les éléments ou la source de données. Ceci est particulièrement vrai pour les Stream parallèles

Important :la modification de la source de données durant l'exécution du pipeline d'opérations peut conduire à un comportement ou un résultat non désiré.

Le type de la valeur de retour de l'opération map() peut être du même type que la valeur traitée ou d'un type différent.

    Stream<String> chaines = Stream.of("aaa", "bbb", "ccc");
    chaines.map(chaine -> chaine.toUpperCase())
           .forEach(System.out::println);

AAA
BBB
CCC

Les traitements de transformations peuvent être le résultat de l'exécution d'une méthode indiquée grâce à une référence de méthode comme pour toute expression Lambda qui ne consiste qu'à invoquer une méthode.

    Stream<String> chaines = Stream.of("aaa", "bbb", "ccc");
    chaines.map(String::toUpperCase)
           .forEach(System.out::println);

AAA
BBB
CCC

Cette méthode peut aussi être une méthode contenant des traitements métiers de l'application.

    Long[] idEmployes = { 12345L, 67890L };
    List<Employe> listeEmployes = Stream
        .of(idEmployes)
        .map(EmployeeService::rechercherParId)
        .collect(Collectors.toList());
    System.out.println(listeEmployes);

package fr.jmdoudoux.dej.streams;

import java.util.Random;

public class EmployeeService {

  private static Random rnd = new Random();

  public static Employe rechercherParId(Long id) {
    return new Employe("nom " + id,
       (id % 2) == 0 ? Genre.HOMME : Genre.FEMME, 
       150 + rnd.nextInt(50),
       1000 + rnd.nextInt(2000));
  }
}

[Employe [nom=nom 12345, genre=FEMME, taille=168,
salaire=2573.0], Employe [nom=nom 67890, genre=HOMME, taille=187, salaire=1374.0]]

La Function peut produire des éléments de types différents du type de l'élément traité.

personnes.stream().map(Personne::getNom).forEach(System.out::println);

Dans l'exemple ci-dessus, la méthode stream() renvoie un Stream<Personne> sur lequel la méthode map() est appliquée. La Function qui lui est passée est une invocation de la méthode getNom() : la méthode map() renvoie un Stream<String> dont les éléments sont les noms des personnes. Enfin la méthode forEach() exécute le Consumer qui affiche tous les noms sur la console.

Plusieurs transformations peuvent être appliquées dans le pipeline d'opérations

    Long[] idEmployes = { 12345L, 67890L };
    List<String> listeNomsEmployes = Stream
        .of(idEmployes)
        .map(EmployeeService::rechercherParId)
        .map(e -> e.getNom())
        .collect(Collectors.toList());
    System.out.println(listeNomsEmployes);

[nom 12345, nom 67890]

La méthode mapToInt() applique une Function à chaque élément pour produire un IntStream à la place d'un Stream<Integer>

La méthode mapToLong() est similaire à la méthode mapToInt() mais elle renvoie un LongStream.

La méthode mapToDouble() est similaire à la méthode mapToInt() mais elle renvoie un DoubleStream.

Lorsque le résultat de la transformation renvoie une valeur entière ou flottante, il est possible d'utiliser la méthode map() qui va renvoyer un Stream du type du wrapper de la valeur primitive.

  Double[] valeurs = { 1.0, 2.0, 3.0, 4.0, 5.0 };
  Double[] valeursAuCarre = Stream.of(valeurs).map(v -> v * v).toArray(Double[]::new);
  System.out.println(Arrays.deepToString(valeursAuCarre));

[1.0, 4.0, 9.0, 16.0, 25.0]

Mais l'utilisation des méthodes mapToDouble(), mapToInt() et mapToLong() améliore les performances essentiellement car elles évitent des opérations coûteuses de boxing et unboxing. Ceci est particulièrement vrai pour très grande quantité d'éléments.

  Double[] valeurs = { 1.0, 2.0, 3.0, 4.0, 5.0 };
  double[] valeursAuCarre = Stream.of(valeurs)
                                  .mapToDouble(v -> v * v)
                                  .toArray();
  System.out.println(Arrays.toString(valeursAuCarre));

[1.0, 4.0, 9.0, 16.0, 25.0]

22.3.2. La méthode flatMap()

Les opérations map() et flatMap() servent toutes les deux à réaliser des transformations mais il existe une différence majeure :

• La méthode map() renvoie un seul élément
• La méthode flatMap() renvoie un Stream qui peut contenir zéro, un ou plusieurs éléments

L'opération map() permet de transformer un élément du Stream mais elle possède une limitation : le résultat de la transformation doit obligatoirement produire un unique élément qui sera ajouté au Stream renvoyé.

L'exemple ci-dessous créé une paire de valeur (valeur -1 et valeur) à partir d'une collection de nombres. La valeur retournée est ainsi une List de List d'Integer.

  List<Integer> nombres = Arrays.asList(1, 3, 5, 7, 9);
  List<List<Integer>> tuples =
      nombres.stream()
             .map(nombre -> Arrays.asList(nombre - 1, nombre))
             .collect(Collectors.toList());
  System.out.println(tuples);

[[0, 1], [2, 3], [4, 5], [6, 7], [8, 9]]

Le même exemple que précédemment mais utilisant une opération flatMap() renvoie une List d'Integer dont toutes les valeurs ont été mises à plat.

  List<Integer> nombres = Arrays.asList(1, 3, 5, 7, 9);
  List<Integer> nombresDesTuples =
      nombres.stream()
             .flatMap(nombre -> Arrays.asList(nombre - 1, nombre).stream())
             .collect(Collectors.toList());
  System.out.println(nombresDesTuples);

[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

L'opération flatMap() transforme chaque élément en un Stream d'autres éléments : ainsi chaque élément va être transformé en zéro, un ou plusieurs autres éléments. Le contenu des Streams issus du résultat de la transformation de chaque objet est agrégé pour constituer le Stream retourné par la méthode flatMap().

L'opération flatMap() attend en paramètre une Function qui renvoie un Stream. La Function est appliquée sur chaque élément qui produit chacun un Stream. Le résultat renvoyé par la méthode est un Stream qui est l'agréation de tous les Streams produits par les invocations de la Function.

package fr.jmdoudoux.dej.streams;

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

public class Departement {
  private String nom;
  private List<Etudiant> etudiants = new ArrayList<>();

  public Departement(String nom) {
    super();
    this.nom = nom;
  }

  public String getNom() {
    return nom;
  }

  public void setNom(String nom) {
    this.nom = nom;
  }
  
  public List<Etudiant> getEtudiants() {
    return etudiants;
  }
}

    departements.stream()
                .flatMap(d -> d.getEtudiants().stream())
                .forEach(System.out::println);

L'exemple ci-dessous affiche tous les mots d'un fichier texte.

    try {
      Files.lines(Paths.get("fichier.txt"))
           .map(ligne -> ligne.split("\\s+"))
           .flatMap(Arrays::stream)
           .map(mot -> mot.replaceAll("[^A-Za-z]+", ""))
           .distinct()
           .forEach(System.out::println);
    } catch (IOException e) {
      e.printStackTrace();
    }

La méthode lines() renvoie un Stream<String> où chaque élément est une ligne du fichier. La première méthode map() renvoie un Stream<String[]> où chaque élément est un tableau des mots d'un ligne. La méthode flatMap() renvoie un Stream<String> où chaque élément est un mot du fichier

L'exemple ci-dessous créé un Stream à partir d'un Stream de List en utilisant la méthode flatMap().

package fr.jmdoudoux.dej.streams;

import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;

public class TestFlatMap {

  public static void main(String[] args) {
    Stream<List<String>> listesIngredients = Stream.of(Arrays.asList("carottes",
"poireaux", "navets"), Arrays.asList("eau"), Arrays.asList("sel", "poivre"));
    Stream<String> ingredients = listesIngredients.flatMap(strList -> strList.stream());
    ingredients.forEach(System.out::println);
  }
}

carottes
poireaux
navets
eau
sel
poivre

22.3.3. La méthode filter()

Il est fréquent de vouloir appliquer un filtre sur les éléments d'un Stream pour ne conserver qu'un sous-ensemble de ces éléments, ceux répondant à un certain critère. L'opération filter() permet de ne conserver que les éléments du Stream qui respecte une certaine condition.

L'interface Stream<T> propose la méthode filter() pour filtrer les éléments du stream :

Stream<T> filter(Predicate<? super T> predicate)

La méthode filter() attend en paramètre une interface fonctionnelle de type Predicate : elle permet de filter les éléments du Stream selon la condition fournie en paramètre par l'interface fonctionnelle de type Predicate. Le Predicate est appliqué sur chaque élément du Stream : s'il renvoie true, élément est ajouté dans le Stream de sortie, sinon l'élément est ignoré.

    Listb<String> prenoms = Arrays.asList("andre", "benoit" "albert", "thierry", "alain",
      "jean");
    prenoms.stream()
           .filter(p -> p.startsWith("a"))
           .forEach(System.out::println);

andre
albert
alain

Il est possible d'utiliser plusieurs opérations filter() dans un même pipeline d'opérations.

    List<String> prenoms = Arrays.asList("andre", "benoit","albert", "thierry", "alain",
      "jean");
    prenoms.stream()
           .filter(p -> p.startsWith("a"))
           .filter(p -> p.length() == 5)
           .forEach(System.out::println);

andre
alain

Il est cependant préférable de regrouper les filtres lorsque cela est possible.

    List<String> prenoms = Arrays.asList("andre", "benoit","albert", "thierry", "alain",
      "jean");
    prenoms.stream()
           .filter(p -> p.startsWith("a") && (p.length() == 5))
           .forEach(System.out::println);

andre
alain

22.3.4. La méthode distinct()

L'opération distinct() permet de supprimer les doublons contenus dans un Stream. Les éléments contenus dans le Stream retournés sont donc uniques. La méthode distinct() n'attend aucun paramètre et renvoie un Steam avec les éléments dont les doublons ont été retirés.

package fr.jmdoudoux.dej.streams;
      
import java.io.PrintStream;
import java.util.Arrays;
import java.util.List;

public class TestStream {

  public static void main(String[] args) {
    List<Integer> entiers = Arrays.asList(1, 2, 3, 1, 2, 3);
    entiers.stream()
           .distinct()
           .forEach(System.out::println);
  }
}

1
2
3

La détection des doublons est réalisée en utilisant la méthode equals() des éléments. Il est donc important que celle-ci soit correctement implémentée.

package fr.jmdoudoux.dej.streams;

public class MaClasse {

  private String nom;

  public MaClasse(String nom) {
    this.nom = nom;
  }

  public String getNom() {
    return nom;
  }

  @Override
  public boolean equals(Object obj) {
    if (this == obj) {
      return true;
    }
    if (obj == null) {
      return false;
    }
    if (getClass() != obj.getClass()) {
      return false;
    }
    MaClasse other = (MaClasse) obj;
    if (nom == null) {
      if (other.nom != null) {
        return false;
      }
    } else if (!nom.equals(other.nom)) {
      return false;
    }
    return true;
  }

  // implémentation de la méthode hashCode() en correspondance

  // volontairement non fournie pour illustrer la problématique engendrée


  @Override
  public String toString() {
    return "MaClasse [nom=" + nom + "]";
  }
}

package fr.jmdoudoux.dej.streams;

import java.io.PrintStream;
import java.util.Arrays;
import java.util.List;

public class TestStream {

  public static void main(String[] args) {
    List<MaClasse> maClasses = Arrays.asList(new MaClasse("aaa"), new MaClasse("bbb"), new
MaClasse("aaa"));
    maClasses.stream()
             .distinct()
             .forEach(System.out::println);
  }
}

MaClasse [nom=aaa]
MaClasse [nom=bbb]
MaClasse [nom=aaa]

La méthode distinct() ne donne pas le résultat attendu bien que la méthode equals() soit implémentée. Cela est dû est fait qu'en Java, il faut implémenter la méthode hashCode() en correspondance avec l'implémentation de la redéfinition de la méthode equals().

package fr.jmdoudoux.dej.streams;

public class MaClasse {

  private String nom;

  public MaClasse(String nom) {
    this.nom = nom;
  }

  public String getNom() {
    return nom;
  }

  @Override
  public int hashCode() {
    final int prime = 31;
    int result = 1;
    result = (prime * result) + ((nom == null) ? 0 : nom.hashCode());
    return result;
  }

  @Override
  public boolean equals(Object obj) {
    if (this == obj) {
      return true;
    }
    if (obj == null) {
      return false;
    }
    if (getClass() != obj.getClass()) {
      return false;
    }
    MaClasse other = (MaClasse) obj;
    if (nom == null) {
      if (other.nom != null) {
        return false;
      }
    } else if (!nom.equals(other.nom)) {
      return false;
    }
    return true;
  }

  @Override
  public String toString() {
    return "MaClasse [nom=" + nom + "]";
  }
}

MaClasse [nom=aaa]
MaClasse [nom=bbb]

Cette opération intermédiaire est stateful : elle nécessite de stocker tous les éléments qui seront renvoyés : cela peut donc requérir une quantité de mémoire dépendante du nombre d'éléments traités par le Stream. Tous les éléments du Stream en entrée doivent être traités avant qu'un premier élement ne soit envoyés dans le Stream de retour.

Si le Stream est ordonné, l'élément conservé parmi les doublons est toujours le premier. Si le Stream n'est pas ordonné, l'élément conservé parmi les doublons n'est pas prédictible.

L'utilisation d'un Stream non ordonné ou l'utilisation de la méthode unordered() peut améliorer les performances lors de l'utilisation de distinct() dans un Stream parallèle, sous réserve que le contexte le permette. Si l'ordre doit être préservé, il arrive qu'un Stream séquentiel puisse être plus performant qu'un Stream parallèle équivalent.

22.3.5. La méthode limit()

L'opération intermédiaire limit() permet de limiter le nombre d'éléments contenu dans le Stream retourné. Le nombre d'éléments est passé en paramètre de la méthode limit().

C'est une opération de type short-circuit : dès que le nombre d'éléments contenus dans le Stream retourné est atteint, l'opération met fin à la consommation d'éléments de la source par le Stream.

package fr.jmdoudoux.dej.streams;

import java.io.PrintStream;
import java.util.Arrays;
import java.util.List;

public class TestStream {

  public static void main(String[] args) {
    List<Integer> entiers = Arrays.asList(1, 2, 3, 4, 5, 6);
    entiers.stream()
           .limit(3)
           .forEach(System.out::println);
  }
}

1
2
3

22.3.6. La méthode skip()

L'opération skip() permet d'ignorer un certain nombre d'éléments du Stream. Les premiers éléments du Stream sont ignorés, les autres sont ajoutés dans le Stream retourné par la fonction. Si le nombre d'éléments fournis en paramètre est supérieur ou égal au nombre d'éléments du Stream, alors le Stream retourné est vide.

La méthode skip() attend en paramètre une valeur entière qui correspond au nombre d'éléments à ignorer.

package fr.jmdoudoux.dej.streams;

import java.io.PrintStream;
import java.util.Arrays;
import java.util.List;

public class TestStream {

  public static void main(String[] args) {
    List<Integer> entiers = Arrays.asList(1, 2, 3, 4, 5, 6);
    entiers.stream()
           .skip(3)
           .forEach(System.out::println);
  }
}

4
5
6

22.3.7. La méthode sorted()

L'opération sorted() permet de trier les éléments du Stream : elle renvoie donc un Stream dont les éléments sont triés. Elle utilise un tampon qui doit avoir tous les éléments avant de pouvoir effectuer le tri.

Par défaut, la méthode sorted() sans paramètre utilise l'ordre naturel des éléments.

package fr.jmdoudoux.dej.streams;

import java.util.Arrays;
import java.util.List;

public class TestStream {

  public static void main(String[] args) {
    List<String> prenoms = Arrays.asList("andre", "benoit", "albert", "thierry", "alain");
    prenoms.stream()
           .sorted()
           .forEach(System.out::println);
  }
}

alain
albert
andre
benoit
thierry

Une surcharge de la méthode sorted() attend en paramètre un Comparator qui sera utilisé pour comparer les éléments deux à deux lors du tri.

   List<String> prenoms = Arrays.asList("andre", "benoit","albert", "thierry", "alain");
   prenoms.stream()
          .sorted(Comparator.reverseOrder())
          .forEach(System.out::println);

thierry
benoit
andre
albert
alain

Il est possible de construire des Comparator évolués en utilisant les méthodes static et par défaut de l'interface Comparator.

   List<String> prenoms = Arrays.asList("andre", "benoit", "albert", "thierry", "alain");
   prenoms.stream()
          .sorted(Comparator.comparingInt(String::length)
                            .thenComparing(Comparator.naturalOrder())
                            .reversed())
          .forEach(System.out::println);

thierry
benoit
lbert
andre
alain

22.3.8. La méthode peek()

L'opération peek() renvoie un Stream contenant tous les éléments du Stream courant en appliquant le Consumer fournit en paramètre sur chacun des éléments.

Le but de cette opération est essentiellement le débogage, par exemple pour afficher l'élément en cours de traitement.

   List<String> prenoms = Arrays.asList("andre", "benoit", "albert", "thierry", "alain",
      "jean");
   prenoms.stream()
          .map(p -> {
            afficher("map   ", p);
            return p.toUpperCase();
          })
          .filter(p -> {
             afficher("filter ", p);
             return p.startsWith("A");
          })
          .peek(p -> {
            afficher("limit ", p);
          })
          .limit(2)
          .forEach(System.out::println);

map : andre
filter : ANDRE
limit  : ANDRE
ANDRE
map    : benoit
filter : BENOIT
map    : albert
filter : ALBERT
limit  : ALBERT
ALBERT
ANDRE

Important : le Consumer ne devrait pas modifier l'élément ni la source du Stream.

22.3.9. Les méthodes qui modifient le comportement du Stream

Plusieurs méthodes sont des opérations intermédiaires qui modifient certaines caractéristiques du Stream et donc peuvent avoir un impact sur la manière dont le Stream va traiter les données.

La méthode sequential() permet de demander l'exécution des traitements du Stream en séquentiel dans un seul thread.

La méthode parallel() permet de demander l'exécution des traitements du Stream en parallèle en utilisant plusieurs threads du pool Fork/Join.

Les méthodes sequential() et parallel() sont des opérations intermédiaires : elles permettent donc simplement de configurer les traitements qui seront effectivement exécutés par l'invocation de la méthode terminale. Les traitements ainsi exécutés ne peuvent l'être dans leur intégralité qu'en séquentiel ou en parallèle même si les méthodes sequential() et parallel() sont invoqués dans le Stream.

    Stream.of(1, 3, 4, 2)
          .parallel()
          .sorted()
          .peek((v) -> System.out.println(Thread.currentThread()
              .getName() + " " + v))
          .sequential()
          .forEach((v) -> System.out.println(Thread.currentThread()
             .getName() + " " + v));

main 1
main 1
main 2
main 2
main 3
main 3
main 4
main 4

C'est l'état indiqué par la dernière méthode sequential() ou parallel() qui sera utilisée pour déterminer le mode d'exécution des traitements du Stream.

    Stream.of(1, 3, 4, 2)
          .sequential()
          .sorted()
          .peek((v) -> System.out.println(Thread.currentThread()
              .getName() + " " + v))
          .parallel()
          .forEach((v) -> System.out.println(Thread.currentThread()
              .getName() + " " + v));

ForkJoinPool.commonPool-worker-2
4
ForkJoinPool.commonPool-worker-3
2
main
3
ForkJoinPool.commonPool-worker-1
1
main
3
ForkJoinPool.commonPool-worker-3
2
ForkJoinPool.commonPool-worker-2
4
ForkJoinPool.commonPool-worker-1 1

La méthode unordered() permet de retirer le flag ORDERED pour indiquer au Stream que les données ne sont pas ordonnées.

Le flag ORDERED du Stream permet de préciser que les éléments sont ordonnées : il peut être positionné par la source (par exemple ArrayList ou un tableau) ou par une opération intermédiaire (par exemple sorted()). Une opération terminale peut ignorer le flag ORDERED (par exemple forEach()). La plupart des opérations intermédiaires traitent les éléments en respectant leur ordre.

Lors de l'exécution des opérations en séquentiel, avoir les éléments dans un certain ordre n'a pas d'influence sur les performances : il affecte uniquement le déterminisme du résultat en fournissant toujours le même avec la même source de données. Si les éléments ne sont pas ordonnés, différentes exécutions sur la même source peuvent donner des résultats différents.

Lors de l'exécution des opérations en parallèle, certaines opérations intermédiaires stateful ou terminales peuvent s'exécuter avec de meilleures performances (exemple distinct() ou Collectors.groupingBy()). Dans ce cas, si l'ordre des éléments n'est pas important, il peut être utile d'utiliser la méthode unordered() du Stream.

Par exemple, la méthode limit() est généralement peu coûteuse lorsque les traitements sont effectués en séquentiel mais peut être très coûteuse sur des données ordonnés avec une exécution en parallèle. Si cela n'altère pas les autres opérations, l'utilisation de la méthode unoredred() peut améliorer les performances dans des traitements en parallèle. Si l'ordre des éléments doit être conservé, l'exécution des traitements en séquentiel peut aussi améliorer les performances notamment celles de la méthode limit().

22.4. Les opérations terminales

Les traitements d'un Stream sont démarrés à l'invocation de son unique opération terminale.

Une seule opération terminale ne peut être invoquée sur un même Stream : une fois cela fait, le Stream ne sera plus utilisable.

Il n'est pas possible de réutiliser un Stream une fois que son opération terminale est invoquée. Le Stream est alors considéré comme consommé. Il faut obligatoirement obtenir un nouveau Stream à partir de la source pour chaque traitement. Il n'est possible que d'invoquer une seule fois un pipeline d'opérations sur un Stream sinon une exception de type IllegalStateException est levée.

  String[] fruits = { "orange", "citron", "pamplemousse", "banane","fraise", 
	  "groseille", "raisin", "pomme", "poire", "abricot", "cerise", "peche", "clementine" };

  Stream<String> stream = Stream.of(fruits);
  stream.filter(s -> s.startsWith("p"))
        .forEach(System.out::println);
  try {
    stream.filter(s -> s.startsWith("c"))
          .forEach(System.out::println);
  } catch (IllegalStateException e) {
    e.printStackTrace(System.out);
  }

Cela ne pose pas de soucis pour créer un nouveau Stream à chaque fois puisque le Stream pointe simplement sur sa source de données mais ne les copie pas. Il est par exemple possible d'utiliser un Supplier pour créer des instances d'un Stream avec les mêmes éléments.

  Supplier<Stream<String>> streamSupplier = () -> Stream.of(fruits);
  streamSupplier.get()
                .filter(s -> s.startsWith("p"))
                .forEach(System.out::println);
  streamSupplier.get()
                .filter(s -> s.startsWith("c"))
                .forEach(System.out::println);

Contrairement aux opérations intermédiaires qui renvoient toujours un Stream, les opérations terminales renvoient une valeur qui correspond au résultat de l'exécution du pipeline d'opérations sur les données. Ce résultat peut être :

• Une valeur d'un type primitif
• Un élément ou une instance de type Optional
• Une collection ou un tableau d'éléments
• void

Il est possible qu'il n'y ait pas de résultat à l'issu des traitements d'un Stream. Java 8 propose la classe Optional qui encapsule une valeur ou l'absence de valeur.

Certaines opérations d'un Stream renvoie donc un objet de type java.util.Optional qui permet de préciser s'il y a un résultat ou non. C'est notamment le cas dans les opérations de réduction, des opérations de recherche (findXXX) ou de recherche de correspondance (XXXMatch) sur un Stream vide.

L'API Stream propose plusieurs opérations terminales dont les principales sont :

Certaines de ces méthodes sont de type short-circuiting, par exemple findFirst().

Il est possible d'exporter les éléments d'un Steam dans une collection ou un tableau en utilisant certaines de ses méthodes :

    Stream<Integer> intStream = Stream.of(1,2,3);
    List<Integer> intList = intStream.collect(Collectors.toList());
    System.out.println(intList);
  

    Stream<Integer> intStream = Stream.of(1, 2, 3);
    Integer[] intArray = intStream.toArray(Integer[]::new);
    System.out.println(Arrays.deepToString(intArray));
  

Une opération de réduction permet de traiter les éléments du Stream de manière itérative pour produire un résultat unique. Des opérations de réduction typique sur des entiers sont par exemple le calcul de la somme ou de la moyenne ou bien encore la détermination de la plus petite ou la plus grande valeur.

Une opération de réduction utilise une valeur initiale (identity) pour la combiner avec le premier élément. Le résultat est ensuite combiné avec le second élément et ainsi de suite. Les traitements utilisés pour réaliser la combinaison sont fournis sous la forme d'une fonction (accumulator).

L'application répétée de cette combinaison permet de générer le résultat qui peut prendre plusieurs formes :

• Un résultat unique (la somme, la moyenne, la plus petite/grande valeur, ...)
• Une collection (List, Set, ...) qui contient les éléments
• Une Map qui contient des paires clé/valeur extraites des données du Stream

L'API Stream propose des opérations de réduction :

• Spécialisée : count(), max(), min(), ...
• Générique : reduce(), collect()

22.4.1. Les méthodes forEach() et forEachOrdered()

L'API Stream propose deux opérations pour exécuter des traitements sur les éléments du résultat de l'exécution des opérations intermédiaires :

• void forEach(Consumer<? super T> action) : cette opération exécute l'action passée en paramètre sur chacun des éléments du Stream. Le comportement de la méthode forEach() sur un Stream parallèle n'est pas déterministe : elle ne garantit pas dans ce cas le respect de l'ordre des éléments puisque ceux-ci sont traités par différents threads
• void forEachOrdered(Consumer<? super T> action) : cette opération est similaire à l'opération forEach() mais elle garantit l'ordre des éléments du Stream. Pour cela, elle n'utilise qu'un seul thread pour exécuter l'action sur tous les éléments. C'est l'API qui détermine quel unique thread exécute l'action sur chacun des éléments

Ces deux méthodes attendent en paramètre un java.util.function.Consumer qui contient l'action à réaliser sur chacun des éléments.

Remarque : il n'est pas possible d'utiliser les instructions break ou return dans l'expression lambda pour interrompre l'itération sur les éléments.

L'exemple ci-dessous illustre les différences lors de l'utilisation de ces méthodes :

package fr.jmdoudoux.dej.streams;

import java.util.Arrays;
import java.util.List;
import java.util.function.Consumer;

public class TestStream {

  public static void main(String[] args) {
    List<String> prenoms = Arrays.asList("andre", "benoit", "albert", "thierry", "alain");
    Consumer<String> afficherElement = s -> System.out.println(s + " - " +
      Thread.currentThread().getName());
    prenoms.stream().sorted().forEach(afficherElement);
    System.out.println();
    prenoms.parallelStream().sorted().forEach(afficherElement);
    System.out.println();
    prenoms.parallelStream().sorted().forEachOrdered(afficherElement);
  }
}

alain - main
albert - main
andre - main
benoit - main
thierry - main
andre - main
thierry - ForkJoinPool.commonPool-worker-3
benoit - main
alain - ForkJoinPool.commonPool-worker-1
albert - ForkJoinPool.commonPool-worker-2
alain - ForkJoinPool.commonPool-worker-2
albert - ForkJoinPool.commonPool-worker-3
andre - ForkJoinPool.commonPool-worker-3
benoit - ForkJoinPool.commonPool-worker-3
thierry - ForkJoinPool.commonPool-worker-3

Lors de l'invocation de la méthode forEach() sur un Stream séquentiel, l'ordre des éléments n'est préservé. Bien que la méthode forEach() ne respecte aucun ordre, le parcours d'un Stream séquentiel se fait dans un unique thread ce qui permet de conserver l'ordre des éléments.

Lors de l'invocation de la méthode forEach() sur un Stream parallèle, l'ordre des éléments est pas préservé. Ceci est lié au fait de l'utilisation de plusieurs threads pour exécuter l'action sur les différents éléments.

Pour respecter l'ordre des éléments dans un Stream parallèle, il est nécessaire d'utiliser la méthode forEachOrdered() qui va réaliser l'exécution de l'action dans un thread unique. Evidemment les performances de la méthode forEachOrdered() sont moins bonnes que celles de la méthode forEach().

Il faut éviter d'utiliser ces méthodes. C'est particulièrement vrai si celles-ci effectuent des traitements qui modifient une donnée.

package fr.jmdoudoux.dej.streams;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.DoubleAdder;

public class TestEmployeStream {

  public static void main(String[] args) {
    List<Employe> employes = new ArrayList<>(6);
    employes.add(new Employe("e1", Genre.HOMME, 176, 1500));
    employes.add(new Employe("e2", Genre.HOMME, 190, 2700));
    employes.add(new Employe("e3", Genre.FEMME, 172, 1850));
    employes.add(new Employe("e4", Genre.FEMME, 162, 3300));
    employes.add(new Employe("e5", Genre.HOMME, 176, 1280));
    employes.add(new Employe("e6", Genre.FEMME, 168, 2850));
    
    DoubleAdder total = new DoubleAdder();
    employes.stream()
            .forEach(e -> total.add(e.getSalaire()));
    
    System.out.println("total salaire = " + total.doubleValue());
  }
}

total salaire = 13480.0

Attention : lors de l'utilisation sur un Stream parallèle, si l'action d'une opération forEach() modifie une ressource partagée, il est nécessaire de gérer explicitement les accès concurrents.

Cette approche n'est cependant pas fonctionnelle : il préférable d'utiliser une approche de type map / reduce. L'interface DoubleStream propose d'ailleurs une méthode sum() qui effectue ce traitement.

package fr.jmdoudoux.dej.streams;

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

public class TestEmployeStream {

  public static void main(String[] args) {
    List<Employe> employes = new ArrayList<>(6);
    employes.add(new Employe("e1", Genre.HOMME, 176, 1500));
    employes.add(new Employe("e2", Genre.HOMME, 190, 2700));
    employes.add(new Employe("e3", Genre.FEMME, 172, 1850));
    employes.add(new Employe("e4", Genre.FEMME, 162, 3300));
    employes.add(new Employe("e5", Genre.HOMME, 176, 1280));
    employes.add(new Employe("e6", Genre.FEMME, 168, 2850));

    Double total = employes.stream()
                           .mapToDouble(e -> e.getSalaire())
                           .sum();

    System.out.println("total salaire = " + total);
  }
}

total salaire = 13480.0

Ces méthodes impliquent nécessairement des effets de bords puisqu'elle ne renvoie rien (void).

Pour respecter une approche fonctionnelle, il est préférable d'éviter de modifier la source de données ou ses éléments.

package fr.jmdoudoux.dej.streams;
      
import java.util.ArrayList;
import java.util.List;

public class TestEmployeStream {

  public static void main(String[] args) {
    List<Employe> employes = new ArrayList<>(6);
    employes.add(new Employe("e1", Genre.HOMME, 176, 1500));
    employes.add(new Employe("e2", Genre.HOMME, 190, 2700));
    employes.add(new Employe("e3", Genre.FEMME, 172, 1850));
    employes.add(new Employe("e4", Genre.FEMME, 162, 3300));
    employes.add(new Employe("e5", Genre.HOMME, 176, 1280));
    employes.add(new Employe("e6", Genre.FEMME, 168, 2850));

    employes.stream()
            .forEach(e -> e.setSalaire(e.getSalaire() + (e.getSalaire() * 0.1)));
    employes.stream()
            .forEach(System.out::println);
  }
}

Employe [nom=e1, genre=HOMME, taille=176, salaire=1650.0]
Employe [nom=e2, genre=HOMME, taille=190, salaire=2970.0]
Employe [nom=e3, genre=FEMME, taille=172, salaire=2035.0]
Employe [nom=e4, genre=FEMME, taille=162, salaire=3630.0]
Employe [nom=e5, genre=HOMME, taille=176, salaire=1408.0]
Employe [nom=e6, genre=FEMME, taille=168, salaire=3135.0]

Une meilleure approche dans ce cas, pourrait être de ne pas utiliser un Stream mais d'utiliser la méthode forEach() de la collection.

package fr.jmdoudoux.dej.streams;

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

public class TestEmployeStream {

  public static void main(String[] args) {
    List<Employe> employes = new ArrayList<>(6);
    employes.add(new Employe("e1", Genre.HOMME, 176, 1500));
    employes.add(new Employe("e2", Genre.HOMME, 190, 2700));
    employes.add(new Employe("e3", Genre.FEMME, 172, 1850));
    employes.add(new Employe("e4", Genre.FEMME, 162, 3300));
    employes.add(new Employe("e5", Genre.HOMME, 176, 1280));
    employes.add(new Employe("e6", Genre.FEMME, 168, 2850));
    
    employes.forEach(e -> { e.setSalaire(e.getSalaire() + (e.getSalaire() * 0.1));   
                               System.out.println(e);
                    });
  }
}

Le résultat de l'exécution est le même que pour l'exemple précédent.

Les méthodes forEach() et forEachOrdered() sont des opérations terminales : une fois leur exécution terminée, le Stream est consommé et ne peux plus être utilisé. Il n'est donc pas possible d'invoquer deux fois ces méthodes sur un même Stream.

package fr.jmdoudoux.dej.streams;

import java.util.stream.Stream;

public class TestStream {

  public static void main(String[] args) {
    Stream<Integer> nombres = Stream.of(1, 2, 3, 4, 5);
    nombres.forEach(i -> System.out.println(i));
    nombres.forEach(System.out::println);
  }
}

1
2
3
4
5
Exception
in thread "main" java.lang.IllegalStateException: stream has already
been operated upon or closed
      at java.util.stream.AbstractPipeline.sourceStageSpliterator(AbstractPipeline.java:279)
      at
java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:580)
      at
fr.jmdoudoux.dej.streams.TestStream.main(TestStream.java:22)

Dans ce cas, il est possible de combiner les deux traitements dans la même expression lambda ou utiliser la méthode peek().

package fr.jmdoudoux.dej.streams;

import java.util.stream.Stream;

public class TestStream {

  public static void main(String[] args) {
    Stream<Integer> nombres = Stream.of(1, 2, 3, 4, 5);

    nombres.peek(i -> System.out.println(i))
           .forEach(System.out::println);
  }
}

1
1
2
2
3
3
4
4
5
5

22.4.2. La méthode collect()

Les opérations de réductions ne renvoient pas toujours un résultat sous la forme d'une valeur unique mais sous la forme d'une structure de données comme une collection par exemple. Dans ce cas, il faut utiliser l'opération collect() qui permet de réaliser une réduction dans un conteneur mutable.

La méthode collect() de l'interface Stream est une opération terminale qui permet de réaliser une agrégation mutable des éléments. Cette agrégation peut prendre différentes formes : stocker les éléments dans une structure de données telle qu'une collection, concaténer les éléments qui sont des chaînes de caractères, ...

L'opération collect() permet de transformer les éléments d'un Stream pour produire un résultat sous différentes formes telles que List, Set, Map, ... ou plus généralement sous la forme d'une réduction dans un conteneur mutable.

Les traitements réalisés par la méthode collect() peuvent être exécutés en parallèle sur plusieurs threads dans un Stream parallèle car chacun d'eux va utiliser sa propre instance du conteneur et la méthode de combinaison permet de fusionner le contenu des différents conteneurs.

La surcharge <R> collect(Supplier<R> resultSupplier, BiConsumer<R, T> accumulator, BiConsumer<R, R> combiner) de la méthode collect() attend trois paramètres :

• supplier : pour fournir une instance du conteneur vide
• accumulator : pour ajouter un élément dans le conteneur
• combiner : pour combiner deux conteneurs lors de l'utilisation du Stream en parallèle

L'exemple ci-dessous utilise la méthode collect() pour mettre tous les éléments du Stream dans une collection de type HashSet grâce aux trois fonctions fournies en paramètre :

• supplier : renvoie une nouvelle instance de type HashSet
• accumulator : ajoute l'élément dans la collection
• combiner : ajoute tous les éléments d'un HashSet dans l'autre

package fr.jmdoudoux.dej.streams;

import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

public class TestCollectors {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem1", "elem2", "elem2", "elem3", "elem4");
    Set<String> ensemble = elements.stream()
                                   .collect(() -> new HashSet<String>(),
                                           (s, e) -> s.add(e),
                                           (s1, s2) -> s1.addAll(s2));
    System.out.println(ensemble);
  }
}

[elem1, elem2, elem3, elem4]

Evidemment, comme les expressions correspondent toutes à l'invocation d'une seule méthode, il est possible d'utiliser des références de méthodes pour obtenir le même résultat.

package fr.jmdoudoux.dej.streams;

import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

public class TestCollectors {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem1", "elem2", "elem2", "elem3", "elem");
    Set<String> ensemble = elements.stream()
                                  .collect(HashSet::new, 
                                            HashSet::add,
                                            HashSet::addAll);
    System.out.println(ensemble);
  }
}

[elem1, elem2, elem3, elem4]

La mise en oeuvre de ces deux exemples est préférable à l'exemple ci-dessous.

package fr.jmdoudoux.dej.streams;
      
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

public class TestCollectors {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem1", "elem2", "elem2", "elem3", "elem4");
    // NE PAS FAIRE COMME CELA

    Set<String> ensemble = new HashSet<>();
    elements.stream()
            .forEach(s -> ensemble.add(s));
    System.out.println(ensemble);
  }
}

[elem1, elem2, elem3, elem4]

Cet exemple fonctionne mais c'est un antipattern. Il fonctionne correctement uniquement en mono thread.

La méthode collect() peut aussi être utilisée pour effectuer une opération de réduction qui va permettre de combiner deux éléments pour obtenir un nouvel élément qui sera combiné à son tour à l'élément suivant et ainsi de suite jusqu'à ce que tous les éléments soient traités.

L'exemple ci-dessous concatène les éléments qui sont de type chaîne de caractères.

package fr.jmdoudoux.dej.streams;
      
import java.util.Arrays;
import java.util.List;

public class TestCollectors {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem1", "elem2", "elem2", "elem3", "elem4");
    StringBuilder elmtSb = 
      elements.stream()
              .collect(() -> new StringBuilder(), 
                       (sb, s) -> sb.append(s), 
                       (sb1, sb2) -> sb1.append(sb2));
    System.out.println(elmtSb.toString());
  }
}

elem1elem2elem2elem3elem4

Comme chaque fonction est l'invocation d'une seule méthode, il est possible d'utiliser des références de méthodes.

package fr.jmdoudoux.dej.streams;

import java.util.Arrays;
import java.util.List;

public class TestCollectors {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem1", "elem2", "elem2", "elem3", "elem4");
    StringBuilder elmtSb = 
      elements.stream()
              .collect(StringBuilder::new, StringBuilder::append, StringBuilder::append);
    System.out.println(elmtSb.toString());
  }
}

elem1elem2elem2elem3elem4

Fréquemment, il est nécessaire d'ajouter un séparateur entre chacun des éléments concaténés.

package fr.jmdoudoux.dej.streams;
      
import java.util.Arrays;
import java.util.List;

public class TestCollectors {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem1", "elem2", "elem2", "elem3", "elem4");
    StringBuilder concat = 
      elements.stream()
              .collect(() -> new StringBuilder(), (sb, s) -> sb.append(";")
              .append(s),(sb1, sb2) -> sb1.append(sb2));
    System.out.println(concat.toString());
  }
}

;elem1;elem2;elem2;elem3;elem4

Evidemment cette solution simple possède l'inconvénient d'avoir le séparateur en trop en première ou dernière position selon l'ordre de concaténation du séparateur et de l'élément courant. Il est alors possible d'écrire un bloc de code pour conditionner l'ajouter du séparateur ou non.

package fr.jmdoudoux.dej.streams;
      
import java.util.Arrays;
import java.util.List;

public class TestCollectors {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem1", "elem2", "elem2", "elem3", "elem4");
    StringBuilder concat = 
      elements.stream()
              .collect(() -> new StringBuilder(), (sb, s) -> {
                       if (sb.length() != 0) {
                         sb.append(";");
                       }
                         sb.append(s);
                       } , (sb1, sb2) -> sb1.append(sb2));
    System.out.println(concat.toString());
  }
}

elem1;elem2;elem2;elem3;elem4

Cette solution utilise un bloc code ce qui la rend plus complexe et empêche l'utilisation d'une référence de méthode. Pour simplifier le code, il est possible d'utiliser la classe StringJoiner introduite dans Java 8.

package fr.jmdoudoux.dej.streams;
      
import java.util.Arrays;
import java.util.List;
import java.util.StringJoiner;

public class TestCollectors {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem1", "elem2", "elem2", "elem3", "elem4");
    StringJoiner elmtJoiner = 
      elements.stream()
              .collect(() -> new StringJoiner(";"),
                       (j, e) -> j.add(e),
                       (j1, j2) -> j1.merge(j2));
    System.out.println(elmtJoiner.toString());
  }
}

elem1;elem2;elem2;elem3;elem4

22.4.3. Les méthodes findFirst() et findAny(

L'interface Stream possède les méthodes findFirst() et findAny() pour obtenir un élément du Stream qui respecte le Predicate qui leur est fourni en paramètre.

Les méthodes findFirst() et findAny() sont des opérations terminales de type short-circuiting qui renvoient une instance de type Optional car il est possible que le Stream ne possède aucun élément et dans ce cas l'instance retournée encapsule l'absence de valeur.

La méthode findFirst() renvoie un objet de type Optional qui encapsule le premier élément dont le Predicate fourni en paramètre est validé.

Si aucun élément n'est trouvé, l'Optional retourné est vide.

package fr.jmdoudoux.dej.streams;
      
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

public class TestFindFirst {

  public static void main(String[] args) {
    List<Personne> personnes = new ArrayList<>(6);
    personnes.add(new Personne("p1", Genre.HOMME, 176));
    personnes.add(new Personne("p2", Genre.HOMME, 190));
    personnes.add(new Personne("p3", Genre.FEMME, 172));
    personnes.add(new Personne("p4", Genre.FEMME, 162));
    personnes.add(new Personne("p5", Genre.HOMME, 176));
    personnes.add(new Personne("p6", Genre.FEMME, 168));

    Optional<Personne> uneGrandePersonne = 
        personnes.stream()
                 .filter(p -> p.getTaille() >= 250)
                 .findFirst();

    if (uneGrandePersonne.isPresent()) {
      System.out.println("Grande personne trouvee : " + uneGrandePersonne);
    } else {
      System.out.println("Aucune grande personne trouvee");
    }
  }
}

Aucune grande personne trouvee

L'opération findAny() renvoie n'importe quel élément du Stream dont le Predicate fourni en paramètre est validé.

L'utilisation de l'une ou l'autre de ces méthodes requière une attention particulière sur un Stream en parallèle.

La méthode findAny() est plus performante que la méthode findFirst() notamment lorsque le Stream est traité en mode parallèle.

22.4.4. Les méthodes xxxMatch()

L'API propose plusieurs méthodes xxxMatch() pour déterminer si aucun, au moins un ou tous les éléments respectent une certaine condition.

La méthode anyMatch(Predicate<? super T> predicate) renvoie un booléen qui précise si au moins un élément du Stream respecte le Predicate.

    Stream<Integer> entiers = Stream.of(1, 2, 3, 4, 5);
    boolean auMoinsUnEgalATrois = entiers.anyMatch(e -> e == 3);
    System.out.println(auMoinsUnEgalATrois);

true

La méthode allMatch(Predicate<? super T> predicate) renvoie un booléen qui précise si tous les éléments du Stream respectent le Predicate. 

  Stream<Integer> entiers = Stream.of(1, 2, 3, 4, 5);
  boolean tousInferieursADix = entiers.allMatch(e -> e < 10);
  System.out.println(tousInferieursADix);

true

La méthode noneMatch(Predicate<? super T> predicate) renvoie un booléen qui précise si aucun élément du Stream respecte le Predicate.

    Stream<Integer> entiers = Stream.of(1, 2, 3, 4, 5);
    boolean tousDifferentsDeDix = entiers.noneMatch(e -> e == 10);
    System.out.println(tousDifferentsDeDix);

true

22.4.5. La méthode count()

La méthode count() renvoie un entier long qui est le nombre d'éléments contenu dans le Stream.

package fr.jmdoudoux.dej.streams;

import java.util.stream.Stream;

public class TestStream {

  public static void main(String[] args) {
    Stream<Integer> nombres = Stream.of(1, 2, 3, 4, 5);
    System.out.println("Nb elements=" + nombres.count());
  }
}

Nb elements=5

La méthode count() est une opération terminale donc elle effectue ces traitements sur le Stream issu de l'exécution des opérations intermédiaires.

package fr.jmdoudoux.dej.streams;

import java.util.stream.Stream;

public class TestStream {

  public static void main(String[] args) {
    Stream<Integer> nombres = Stream.of(1, 2, 3, 4, 5).filter(e -> (e % 2) == 0);
    System.out.println("Nb elements=" + nombres.count());
  }
}

Nb elements=2

22.4.6. La méthode reduce

La réduction, aussi appelée folding en programmation fonctionnelle, permet de réaliser une opération d'agrégation sur un ensemble d'éléments afin de produire un résultat. Une opération de réduction applique de manière répétée une opération sur chacun des éléments pour les combiner afin de produire un unique résultat.

Par exemple, pour des éléments de type entier, si l'opérateur binaire fait une addition alors l'opération de réduction calcule la somme des éléments. Mais les traitements d'une réduction ne se limite pas à ce type de traitements. Une opération de réduction peut aussi être la recherche de la plus grande valeur en utilisant une expression lambda (x,y) -> Math.max(x,y) ou avec la référence de méthode équivalente Math::max.

Une réduction peut aussi s'appliquer sur tout type d'objet : dans ce cas, il est nécessaire de préciser le comportement des traitements de l'opération d'agrégation.

La méthode reduce() effectue une opération de type réduction. L'interface Stream<T> possède trois surcharges de la méthode reduce() :

La surcharge qui attend en paramètre un BinaryOperator<T> renvoie comme résultat au plus un élément encapsulé dans un Optional. Elle attend en paramètre un opérateur binaire associatif qui effectue la réduction des éléments du Stream. Elle renvoie un Optional qui est vide si le Stream ne contient aucun élément. Si le Stream ne possède qu'un seul élément, alors c'est cet élément qui est retourné.

    employes.stream()
            .reduce((p1, p2) -> p1.getTaille() > p2.getTaille() ? p1 : p2)
            .ifPresent(System.out::println);

L'exemple ci-dessus permet de déterminer la personne ayant la plus grande taille.

La fonction de type BinaryOperator<T> attend en paramètres deux objets de type T et renvoie un objet de T. Le premier paramètre est la valeur accumulée et le second paramètre est l'élément courant en cours de traitement dans le Stream.

    String lettres = Stream.of("a", "b", "c", "d")
                           .reduce((accumulator, item) -> accumulator + ", " + item)
                           .orElse("");
    System.out.println(lettres);

a, b, c, d

L'opération de réduction peut utiliser un des BinaryOperator fournis par le JDK. L'exemple ci-dessous recherche la personne la plus grande.

    Comparator<Employe> comparateurParTaille = Comparator.comparingInt(Employe::getTaille);
    BinaryOperator<Employe> lePlusGrand = BinaryOperator.maxBy(comparateurParTaille);
    Optional<Employe> plusGrandEmploye = employes.stream()
                                                 .reduce(lePlusGrand);
    System.out.println(plusGrandEmploye);

La seconde surcharge de la méthode reduce() attend en paramètre la valeur initiale et une fonction de type BinaryOperator qui sera invoquée pour effectuer la réduction.

package fr.jmdoudoux.dej.streams;

import java.io.PrintStream;
import java.util.Arrays;
import java.util.List;

public class TestStream {

  public static void main(String[] args) {
    List<Integer> entiers = Arrays.asList(1, 2, 3, 4, 5, 6);
    Integer total = entiers.stream()
                           .reduce(0, (valeurAccumulee, valeur) -> valeurAccumulee + valeur);
    System.out.println("total=" + total);
  }
}

total=21

La fonction de type BinaryOperator peut être fournie sous la forme d'une référence de méthode. L'exemple ci-dessous détermine la valeur maximum d'éléments de type Integer.

    Integer max = Stream.of(1, 2, 3, 4, 5)
                        .reduce(0, Integer::max);
    System.out.println(max);

Il faut prêter attention à la valeur initiale fournie.

    String lettres = Stream.of("a", "b", "c", "d")
                           .reduce("", (accumulator, item) -> accumulator + ", " + item);
    System.out.println(lettres);

, a, b, c, d

Si le Stream ne contient aucun élément, alors c'est la valeur initiale qui est renvoyée.

    String lettres = Stream.of("a", "b", "c", "d")
                           .limit(0)
                           .reduce("X", (accumulator, item) -> accumulator + ", " + item);
    System.out.println(lettres);

X

L'identity est la valeur initiale qui sera fournie en premier paramètre lors de la première invocation de l'opération d'accumulation.

    int produit = Stream.of(1, 2, 3, 4, 5)
                        .reduce(0, Integer::max);
    System.out.println(produit);

5

Il est aussi possible de concaténer les éléments de type chaînes de caractères.

La valeur initiale est simplement retournée si le Stream ne contient aucun élément.

Il est important que l'agrégation de la valeur initiale avec un élément donne l'élément et vice versa que l'agrégation d'un élément avec la valeur initiale donne aussi l'élément.

Toutes les opérations n'ont pas de valeur initiale et lorsqu'elles en possèdent une elles ne permettent pas toujours d'avoir le résultat attendu. Par exemple, pour une opération de réduction qui détermine la plus grande valeur d'un ensemble d'entiers. Il peut sembler raisonnable d'utiliser la valeur Integer.MIN_VALUE comme identity. Cela fonctionne très bien si le Stream contient au moins un élément mais si le Stream ne contient aucun élément, la valeur retournée ne permet pas de déterminer si elle correspond au fait qu'il n'y a aucun élément ou si c'est vraiment la valeur du plus petit élément. C'est la raison pour laquelle l'interface Stream propose une surcharge qui renvoie un Optional et une autre surcharge qui attend en paramètre la valeur initiale.

Pour la concaténation de chaînes de caractères, la valeur initiale est une chaîne vide.

    List<String> chaines = Arrays.asList("1", "2", "3", "4", "5");
    String chaine = chaines.stream().reduce("", String::concat);
    System.out.println(chaine);

12345

Pour la somme de valeurs entières, la valeur initiale est zéro.

   List<Integer> entiers = Arrays.asList(1, 2, 3, 4, 5, 6);
   int somme = entiers.stream().reduce(0, (x, y) -> x + y);
   System.out.println(somme);

21

Remarque : il est préférable dans ce cas d'utiliser l'opération sum() de l'interface IntStream.

Parfois en fonction de l'opération d'agération, il faut prendre des précautions avec la valeur initiale.

    int produit = Stream.of(1, 2, 3, 4, 5)
                        .reduce(0, (a, b) -> a * b);
    System.out.println(produit);

0

Ce n'est pas le résultat attendu car la valeur initiale est passée en premier paramètre de l'opération de réduction. Dans le cas où l'opération est une multiplication, il faut utiliser la valeur 1 comme valeur initiale.

    int produit = Stream.of(1, 2, 3, 4, 5)
                        .reduce(1, (a, b) -> a * b);
    System.out.println(produit);

120

La troisième surcharge de la méthode reduce() attend en paramètre la valeur initiale, une BiFunction qui sera utilisée comme un accumulateur et une BinaryOperator qui sera utilisée comme combinateur.

Cette opération peut être définie explicitement de manière différente au moyen de deux opérations de type map() et reduce().

L'exemple ci-dessous calcule la taille des employés. Ce traitement aurait pu être réalisé différemment par le Stream mais il est utilisé pour illustrer le mode de fonctionnement de la méthode reduce().

   Integer tailleTotale = employes.stream()
      .reduce(0, (somme, e) -> somme += e.getTaille(),
                 (somme1, somme2) -> somme1 + somme2);
  System.out.println("Taille totale = " + tailleTotale);

Taille totale = 1044

Le traitement exécuté est similaire à celui-ci-dessous.

    BiFunction<Integer, Employe, Integer> accumulator = (somme, e) -> somme += e.getTaille();
    Integer resultat = 0;
    for (Employe e : employes) {
      resultat = accumulator.apply(resultat, e);
    }
  System.out.println("Taille totale = " + tailleTotale);

Taille totale = 1044

Pour comprendre le mode de fonction de la méthode reduce() et l'utilisation qu'elle fait des fonctions d'accumulation et de combinaison, il faut ajouter des traces dans leurs expressions Lambdas.

   Integer tailleTotale = employes.stream().reduce(0, (somme, e) -> {
     afficher("accumulator", " somme=" + somme + " e=" + e.getTaille());
     return somme += e.getTaille();
   } , (somme1, somme2) -> {
     afficher("combiner", " somme1=" + somme1 + " somme2=" + somme2);
     return somme1 + somme2;
   });
   System.out.println("Taille totale = " + tailleTotale);

accumulator:
somme=0 e=176 [main]
accumulator:
somme=176 e=190 [main]
accumulator:
somme=366 e=172 [main]
accumulator:
somme=538 e=162 [main]
accumulator:
somme=700 e=176 [main]
accumulator:
somme=876 e=168 [main]
Taille totale = 1044

La fonction de combinaison n'est jamais invoquée lorsque le Stream est exécuté en séquentiel. Le comportement est différent lorsque le Stream est invoqué en parallèle.

   Integer tailleTotale = employes.parallelStream().reduce(0, (somme, e) -> {
     afficher("accumulator", " somme=" + somme + " e=" + e.getTaille());
     return somme += e.getTaille();
   } , (somme1, somme2) -> {
     afficher("combiner", " somme1=" + somme1 + " somme2=" + somme2);
     return somme1 + somme2;
   });
   System.out.println("Taille totale = " + tailleTotale);
 } 

accumulator:  somme=0 e=190
[ForkJoinPool.commonPool-worker-1]
accumulator:  somme=0 e=176
[ForkJoinPool.commonPool-worker-3]
accumulator:  somme=0 e=168
[ForkJoinPool.commonPool-worker-2]
accumulator:  somme=0 e=162 [main]
accumulator:  somme=0 e=176
[ForkJoinPool.commonPool-worker-3]
accumulator:  somme=0 e=172
[ForkJoinPool.commonPool-worker-1]
combiner:  somme1=176 somme2=168
[ForkJoinPool.commonPool-worker-3]
combiner:  somme1=190 somme2=172 
[ForkJoinPool.commonPool-worker-1]
combiner:  somme1=162 somme2=344
[ForkJoinPool.commonPool-worker-3]
combiner:  somme1=176 somme2=362
[ForkJoinPool.commonPool-worker-1]
combiner:  somme1=538 somme2=506
[ForkJoinPool.commonPool-worker-1]
Taille totale = 1044

Le comportement des traitements est différent : comme l'accumulateur est invoqué en parallèle, ce n'est plus lui qui fait la somme mais la fonction de combinaison.

Une opération de réduction est particulièrement intéressante lors du traitement du Stream en parallèle. Lors de l'exécution des traitements en parallèle, il est nécessaire de partager une variable qui stocke la valeur cumulée et de gérer les accès concurrents à cette variable par les différents threads. Généralement, cette gestion des accès concurrents inhibe voire dégrade les performances gagnées par la parallélisation.

Pour limiter cet impact, l'opération reduce() d'un Stream, traite et cumule les éléments dans chaque threads : ainsi, il n'y a pas de variable partagée et donc aucune contention liée aux accès concurrents.

Les résultats de chaque threads sont ensuite combinés pour obtenir le résultat final.

Pour permettre à une opération de réduction d'être exécutée en parallèle, il est nécessaire que celle-ci soit associative.

Une opération # est associative si (a # b) # c = a # (b # c).

L'addition et la multiplication sont des opérations associatives. La soustraction n'est pas une opération associative :

(3 - 2) -1 = 0

3 - (2 -1) = 1

     OptionalInt res = IntStream.range(1, 100)
                                .reduce((a, b) -> a - b);
     System.out.println(res.getAsInt());

-4948

Avec une opération associative, la réduction peut être effectuée dans n'importe quel ordre. Ceci est important pour une exécution en parallèle dans laquelle les éléments sont traités par lots, la réduction se faisant pour chacun des éléments des lots puis il y a une combinaison des résultats intermédiaires pour renvoyer le résultat de la réduction.

Si l'opération n'est pas associative, les résultats obtenus ne seront pas ceux escomptés.

      OptionalInt res = IntStream.range(1, 100)
                                 .parallel()
                                 .reduce((a, b) -> a - b);
      System.out.println(res.getAsInt());

24

L'opération n'a pas besoin d'être commutative. Une opération # est commutative si a # b = b # a. L'addition et la multiplication sont des opérations commutatives. La soustraction n'est pas une opération commutative :

3 - 2 = 1

2 - 3 = -1

Un exemple d'opération de réduction qui est associative mais pas commutative est la concaténation de chaînes de caractères.

22.4.7. Les méthodes min() et max()

Les méthodes min() et max() permettent respectivement de retourner la valeur minimale et maximale issue des traitements du Stream.

Elles attendent en paramètre un Comparator qui permet de préciser l'ordre de comparaison des éléments.

Elles renvoient une instance de type Optional<T>

Le plus simple est d'utiliser la méthode comparing() de l'interface Comparator : elle renvoie un objet de type Comparator qui compare les clés extraites grâce à l'expression Lambda fournie en paramètre. Le paramètre de la méthode comparing() est une Function qui permet d'extraire la clé sur laquelle le Comparator retourné va faire la comparaison.

package fr.jmdoudoux.dej.streams;
      
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.Optional;

public class TestStream {

  public static void main(String[] args) {
    List<String> prenoms = Arrays.asList("andre", "benoit", "albert", "thierry", "alain",
    "jean");
    Optional<String> plusPetitPrenom = 
      prenoms.stream().min(Comparator.comparing(element -> element.length()));
    System.out.println(plusPetitPrenom.orElseGet(() -> "aucun prenom trouve"));
  }
}

jean

22.4.8. La méthode toArray()

La méthode toArray() permet de renvoyer les éléments du Stream dans un tableau. Elle possède deux surcharges :

Si le type du tableau n'est pas important, il est possible d'utiliser la méthode toArray() sans paramètre.

    Stream<String> stream = Stream.of("a", "b", "c");
    Object[] strings = stream.toArray();

La Function attendue en paramètre de la surcharge de la méthode toArray() attend en paramètre un entier qui est la taille du tableau et renvoie un tableau dont la taille est celle passée en paramètre. Elle permet de préciser le type du tableau à utiliser et qui sera renvoyé par la méthode.

Stream<String> stream = Stream.of("a", "b", "c"); 
String[] strings = stream.toArray(size -> new String[size]);

Il est aussi possible d'utiliser une référence de méthode, ce qui est plus simple et limite les possibilités d'erreurs.

Stream<String> stream = Stream.of("a","b" "c");
String[] strings = stream.toArray(String[]::new);

Ce code correspond à celui-ci-dessous qui utilise une classe anonyme interne :

Stream<String> stream = Stream.of("a", "b", "c");
String[] strings = stream.toArray(new IntFunction<String[]>() {
    @Override
	public String[] apply(int > size) {
       return new String[size];       
    }
});

La méthode toArray() est utilisable sur n'importe quel Stream y compris sur les Streams primitifs.

   Integer[] integerArray = Stream.of(1, 2, 3, 4, 5).toArray(Integer[]::new);
   int[] class=n>intArray = IntStream.of(1, 2, 3, 4, 5).toArray();
   long[] longArray = LongStream.of(1, 2, 3, 4, 5).toArray();
   double[] doubleArray = DoubleStream.of(1, 2, 3, 4, 5).toArray();

22.4.9. La méthode iterator

La méthode iterator() de l'interface BaseStream renvoie un Iterator<T> qui permet de parcourir dans une itération extérieure tous les éléments d'un Stream.

    Stream<String> stream = Stream.of("a", "b", "c");
    Iterator<String> it = stream.iterator();
    while (it.hasNext()) {
      String s = it.next();
      System.out.println(s);
    }

Attention, comme un Stream ne stocke pas ses éléments, il n'est possible de parcourir les éléments avec l'Iterator qu'une seule fois.

package fr.jmdoudoux.dej.streams;

import java.util.Iterator;
import java.util.stream.Stream;

public class TestStreamIterator {

  public static void main(String[] args) {
    Stream<String> stream = Stream.of("a", "b", "c");

    Iterator<String> it = stream.iterator();
    while (it.hasNext()) {
      String s = it.next();
      System.out.println(s);
    }

    it = stream.iterator();
    while (it.hasNext()) {
      String s = it.next();
      System.out.println(s);
    }
  }
}

a
b
c
Exception in thread "main"
java.lang.IllegalStateException: stream has already been operated upon or closed
      at java.util.stream.AbstractPipeline.spliterator(AbstractPipeline.java:343)
      at java.util.stream.ReferencePipeline.iterator(ReferencePipeline.java:139)
      at fr.jmdoudoux.dej.streams.TestStreamIterator.main(TestStreamIterator.java:16)

La méthode iterator() renvoie un Iterator. Cependant parfois c'est un Iterable qui est requis notamment pour une utilisation dans une instruction for. Il est alors possible de caster en Iterable une référence de méthode sur la méthode iterator() du Stream.

    Stream<String> stream = Stream.of("a", "b", "c");
    for (String s : (Iterable<String>)stream::iterator) {
      System.out.println(s);
    }

Comme l'interface Iterable ne possède qu'une seule méthode abstraite Iterator<T> iterator(), c'est une interface fonctionnelle. Elle peut donc être implémenté sous la forme d'une expression Lambda et donc d'une référence de méthode. La signature de la méthode iterator() de l'interface Stream correspond à la signature de la méthode iterator() de l'interface Iterable : il est donc possible d'utiliser une référence de méthode sur la méthode iterator() de l'interface Stream pour implémenter l'interface Iterable. Il est cependant nécessaire d'effectuer un cast pour forcer le target type de la référence de méthode.

22.5. Les Collectors

Une des surcharges de la méthode collect() attend en paramètre un objet de type Collector. Les traitements à appliquer sont alors définis par l'interface Collector.

La classe java.util.stream.Collectors propose un ensemble de fabriques qui renvoient des implémentations de Collector pour des opérations de réduction communes.

22.5.1. L'interface Collector

Un Collector permet de réaliser une opération de réduction qui accumule les éléments d'un Stream dans un conteneur mutable. Il peut éventuellement appliquer une transformation pour permettre de fournir le résultat final dans un type différent de celui du conteneur dans lequel les éléments sont accumulés. Les traitements des opérations de réduction peuvent être exécutés de manière séquentielle ou parallèle.

Les traitements d'un Collector sont définis grâce à 4 fonctions qui sont utilisées pour agréger les éléments du Stream dans un conteneur mutable, avec éventuellement une transformation optionnelle pour produire le résultat final :

• Un supplier : permet de renvoyer une nouvelle instance du conteneur mutable
• Un accumulator : accumule un élément dans le conteneur
• Un combiner : combine une instance du type du conteneur pour en produire un seul
• Un finisher : transforme le conteneur pour renvoyer une instance du type du résultat renvoyé (cette fonction est optionnelle)

L'interface Collector<T,A,R> possède trois types génériques :

• T : le type des éléments utilisé par l'opération de réduction
• A : le type de l'objet mutable utilisé par l'opération de réduction
• R : le type du résultat de l'opération de réduction

Elle définit plusieurs méthodes :

L'utilisation d'une implémentation d'un Collector dans des traitements séquentiels réalise basiquement plusieurs traitements :

• Utilise le Supplier pour créer une instance du conteneur
• Invoque la fonction accumulator pour chaque élément
• Invoque éventuellement la fonction finisher si nécessaire à la fin

L'utilisation d'une implémentation d'un Collector dans des traitements parallèles réalise basiquement plusieurs traitements :

• Utilise le Supplier pour créer une instance du conteneur pour chaque lot
• Invoque la fonction accumulator pour chaque élément du lot
• Invoque la fonction combiner pour fusionner les conteneurs de chaque lot
• Invoque éventuellement la fonction finisher si nécessaire à la fin

Un Collector peut avoir un ensemble de caractéristiques qui peuvent être utilisées pour optimiser les traitements et ainsi améliorer les performances. Ces caractéristiques sont stockées dans une collection de type Set immuable de valeurs définies dans l'énumération Collector.Characteristics.

Elles peuvent permettre la mise en oeuvre éventuelle d'optimisations dans le traitement de réduction du Collector.

22.5.2. La classe Collectors

La classe Collectors propose des fabriques pour obtenir des instances de Collector qui réalisent des agrégation communes telles que l'ajout des éléments dans une collection, la concaténation de chaînes de caractères, des réductions, des calculs numériques et statistiques, des groupements, ...

Elle propose des méthodes statiques qui sont des fabriques pour renvoyer des instances de type Collector fournies en standard dans l'API permettant de répondre aux principaux besoins courants :

Le plus simple pour les utiliser est de réaliser un import static

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

Dans l'exemple ci-dessous, la méthode collect() utilise le Collector renvoyé par la méthode toList() de la classe Collectors pour renvoyer une collection de type List contenant les éléments du Stream.

    List<Employe> employesMasculins = employes
        .stream()
        .filter(e -> e.getGenre() == Genre.HOMME)
        .collect(Collectors.toList());
    System.out.println(employesMasculins);

De la même façon, pour obtenir un collection de type Set, il suffit d'utiliser le Collector retourné par la méthode toSet() de la classe Collectors.

    Set<Employe> employesMasculins = employes
      .stream()
      .filter(e -> e.getGenre() == Genre.HOMME)
      .collect(Collectors.toSet());
    System.out.println(employesMasculins);

Il est possible d'utiliser la méthode toCollection() pour pouvoir fournir l'instance de la collection à utiliser pour l'agrégation des éléments. Par exemple, pour agréger les éléments dans une collection de type TreeSet sur des éléments qui nécessairement implémentent l'interface Comparable.

    Set<Employe> employesMasculins = employes
        .stream()
        .filter(p -> p.getGenre() == Genre.HOMME)
       .collect(Collectors.toCollection(TreeSet::new));
    System.out.println(employesMasculins);

Si les éléments n'implémentent pas Comparable, il est possible d'utiliser une expression lambda pour implémenter le Supplier qui va instancier le TreeSet en passant en paramètre de son constructeur le Comparator à utiliser.

    Set<Employe> employesMasculins = employes
       .stream()
       .filter(p -> p.getGenre() == Genre.HOMME)
       .collect(Collectors.toCollection(() -> 
         new TreeSet<Employe>(Comparator.comparing(Employe::getNom))));
    System.out.println(employesMasculins);

Les Collector ne se contentent pas de retourner des collections, ils peuvent aussi réaliser des agrégations pour déterminer un résultat.

Dans l'exemple ci-dessous, la méthode averagingDouble() de la classe Collectors renvoie un Collector qui calcule la moyenne des données obtenues en invoquant la ToDoubleFunction() passée en paramètre sur chacun des éléments.

  Double salaireMoyen = employes.stream()
                                .collect(Collectors.averagingDouble(Employe::getSalaire));
  System.out.println(salaireMoyen);

Certains Collector peuvent être plus complexes. Par exemple, la méthode groupingBy() de la classe Collectors renvoie un Collector qui va grouper dans une Map les éléments sur la base d'une clé précisée.

    Map<Genre, List<Employe>> employesParGenre =
      employes.stream()
              .collect(Collectors.groupingBy(Employe::getGenre));
    employesParGenre.forEach((genre, listeEmployes) -> 
      System.out.format("%s : %s\n", genre, listeEmployes));

FEMME : [Employe [nom=e3, genre=FEMME, taille=172, salaire=1850.0], Employe [nom=e4, genre=
FEMME, taille=162, salaire=3300.0], Employe [nom=e6, genre=FEMME, taille=168, salaire=2850.0]]
HOMME : [Employe [nom=e1, genre=HOMME, taille=176, salaire=1500.0], Employe [nom=e2, genre=
HOMME, taille=190, salaire=2700.0], Employe [nom=e5, genre=HOMME, taille=176, salaire=1280.0]]

22.5.2.1. Les fabriques pour des Collector vers des collections

La méthode toList() permet de renvoyer les éléments du Stream dans une collection de type List.

Remarque : il n'est pas possible avec cette méthode de pouvoir préciser le type de l'implémentation de la collection. Pour cela, il faut utiliser la méthode toCollection().

package fr.jmdoudoux.dej.streams;
      
import static java.util.stream.Collectors.toList;
import java.util.Arrays;
import java.util.List;

public class TestCollector {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem1", "elem2", "elem3", "elem4");
    List<String> resultats = elements.stream().collect(toList());
    System.out.println(resultats);
  }
}

La méthode toSet() permet de renvoyer les éléments du Stream dans une collection de type Set.

package fr.jmdoudoux.dej.streams;

import static java.util.stream.Collectors.toSet;
import java.util.Arrays;
import java.util.List;
import java.util.Set;

public class TestCollector {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem1", "elem2", "elem2", "elem3", "elem4");
    Set<String> resultats = elements.stream().collect(toSet());
    System.out.println(resultats);
  }
}

Il n'y a aucune garantie sur l'instance de type Set retournée par le Collector de la méthode toSet().

Remarque : il n'est donc pas possible avec cette méthode de pouvoir préciser le type de l'implémentation de la collection. Pour cela, il faut utiliser la méthode toCollection().

La méthode toCollection() permet de renvoyer les éléments du Stream dans une collection dont l'implémentation est fournie par le Supplier.

package fr.jmdoudoux.dej.streams;

import static java.util.stream.Collectors.toCollection;
import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.TreeSet;

public class TestCollectors {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem4", "elem2", "elem2", "elem1", "elem3");
    Set<String> resultats = elements.stream().collect(toCollection(TreeSet::new));
    System.out.println(resultats);
  }
}

Remarque : l'implémentation de type Collection fournie doit être mutable

L'utilisation permet d'avoir un contrôle précis sur l'instance utilisée. L'exemple ci-dessous est le même que le précédent mais l'instance de type Set utilisée est différente.

package fr.jmdoudoux.dej.streams;

import static java.util.stream.Collectors.toCollection;
import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.TreeSet;

public class TestCollectors {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem4", "elem2", "elem2", "elem1", "elem3");
    Set<String> resultats = elements.stream().collect(toCollection(HashSet::new));
    System.out.println(resultats);
  }
}

Les surcharges de la méthode toMap() permettent de renvoyer les éléments du Stream dans une collection de type Map.

package fr.jmdoudoux.dej.streams;

import static java.util.stream.Collectors.toMap;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.function.Function;

public class TestCollectors {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem1", "elem2", "elem3", "elem4");
    Map<String, Integer> resultats = elements.stream().collect(toMap(Function.identity(), 
                                                                           String::length));
    System.out.println(resultats);
  }
}

{elem1=8, elem2=8, elem3=8, elem4=8}

Remarque : cette méthode ne permet pas d'avoir un support des clés dupliquées.

package fr.jmdoudoux.dej.streams;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;

public class TestCollectors {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem1", "elem2", "elem3", "elem4");
    Map<Integer, String> resultats = 
      elements.stream()
              .collect(Collectors.toMap(String::length, Function.identity()));
    System.out.println(resultats);
  }
}

Exception in thread "main" java.lang.IllegalStateException: Duplicate key elem1
      at java.util.stream.Collectors.lambda$throwingMerger$114(Collectors.java:133)
      at java.util.HashMap.merge(HashMap.java:1245)
      at java.util.stream.Collectors.lambda$toMap$172(Collectors.java:1320)
      at java.util.stream.ReduceOps$3ReducingSink.accept(ReduceOps.java:169)
      at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
      at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
      at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
      at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708)
      at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
      at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499)
      at fr.jmdoudoux.dej.streams.TestCollectors.main(TestCollectors.java:17)

La surcharge toMap(Function<? super T,? extends K> keyMapper, Function<? super T,? extends U> valueMapper, BinaryOperator<U> mergeFunction) permet de renvoyer les éléments du Stream dans une collection de type Map en tenant compte des clés dupliquées grâce à la fonction de fusion fournie sous la forme d'un BinaryOperator (mergeFunction).

package fr.jmdoudoux.dej.streams;

import static java.util.stream.Collectors.toMap;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.function.Function;

public class TestCollectors {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem1", "elem2", "elem2", "elem3", "elem4");
    Map<Integer, String> resultats = 
      elements.stream().collect(toMap(String::length, Function.identity(), (s1, s2) -> s1));
    System.out.println(resultats);
  }
}

{8=elem1}

Dans l'exemple ci-dessus, comme les clés sont identiques, la fonction de fusion des doublons renvoie simplement le premier des deux. Evidemment, la fonction peut être adaptée selon les besoins, par exemple pour cumuler les valeurs des différents éléments de chaque clé.

package fr.jmdoudoux.dej.streams;

import static java.util.stream.Collectors.toMap;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.function.Function;

public class TestCollectors {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem1", "elem2", "elem2", "elem3", "elem4");
    Map<Integer, String> resultats = 
      elements.stream()
              .collect(toMap(String::length, 
                             Function.identity(),
                             (s1, s2) -> s1 + ";" + s2));
    System.out.println(resultats);
  }
}

{8=elem1;elem2;elem2;elem3;elem4}

22.5.2.2. La fabrique pour des Collector qui exécutent une action complémentaire

La méthode collectingAndThen() permet d'exécuter une action supplémentaire sous la forme d'une fonction après l'exécution du Collector fourni en paramètre.

package fr.jmdoudoux.dej.streams;

import static java.util.stream.Collectors.collectingAndThen;
import static java.util.stream.Collectors.toList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;

public class TestCollector {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem1", "elem2", "elem3", "elem4");
    List<String> resultats = elements.stream()
      .collect(collectingAndThen(toList(), Collections::unmodifiableList));
    System.out.println(resultats.getClass().getName());
  }
}

22.5.2.3. Les fabriques qui renvoient des Collector pour réaliser une agrégation

Les surcharges de la méthode joining() de la classe Collectors permettent de concaténer les éléments d'un Stream<String>.

La surcharge sans paramètre joining() renvoie un Collector qui permet de concaténer les éléments d'un Steam<String>

package fr.jmdoudoux.dej.streams;

import static java.util.stream.Collectors.joining;
import java.util.Arrays;
import java.util.List;

public class TestCollector {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem1", "elem2", "elem3", "elem4");
    String resultat = elements.stream().collect(joining());
    System.out.println(resultat);
  }
}

La surcharge joining(CharSequence delimiter) renvoie un Collector qui permet de concaténer les éléments d'un Stream<String> avec le séparateur fourni en paramètre.

package fr.jmdoudoux.dej.streams;

import static java.util.stream.Collectors.joining;
import java.util.Arrays;
import java.util.List;

public class TestCollector {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem1", "elem2", "elem3", "elem4");
    String resultat = elements.stream().collect(joining(","));
    System.out.println(resultat);
  }
}

La surcharge joining(CharSequence delimiter, CharSequence prefix, CharSequence suffix) renvoie un Collector qui permet de concaténer les éléments d'une Steam<String> avec le séparateur, le préfixe et le suffixe fournis en paramètre.

package fr.jmdoudoux.dej.streams;

import static java.util.stream.Collectors.joining;
import java.util.Arrays;
import java.util.List;

public class TestCollector {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem1", "elem2", "elem3", "elem4");
    String resultat = elements.stream().collect(joining(", ", "[", "]"));
    System.out.println(resultat);
  }
}

Le préfixe et le suffixe peuvent être n'importe quelle chaîne de caractères.

    String noms = employes
        .stream()
        .map(Employe::getNom)
        .collect(Collectors.joining(", ", "Les personnes ", " sont des employes."));
    System.out.println(noms);

Les personnes e1, e2, e3, e4, e5, e6 sont des employes.

22.5.2.4. Les fabriques qui renvoient des Collectors pour effectuer des opérations numériques

La méthode counting() renvoie un Collector qui permet de compter le nombre d'éléments du Stream.

package fr.jmdoudoux.dej.streams;

import static java.util.stream.Collectors.counting;
import java.util.Arrays;
import java.util.List;

public class TestCollector {

  public static void main(String[] args) {
    List<String> elements = Arrays.asList("elem1", "elem2", "elem3", "elem4");
    Long resultat = elements.stream().collect(counting());
    System.out.println(resultat);
  }
}

Les méthodes summarizingInt(), summarizingLong() et summarizingDouble() de la classe Collectors renvoient des Collector qui calculent des informations statistiques basiques sur les données numériques extraites des éléments du Stream : le nombre d'éléments, les valeurs min et max, la moyenne et la somme.

Ces données sont respectivement encapsulées dans des objets de type IntSummaryStatistics, LongSummaryStatistics, et DoubleSummaryStatistics.

    DoubleSummaryStatistics salaireSummary = employes
      .stream()
      .collect(Collectors.summarizingDouble(Employe::getSalaire));
    System.out.println(salaireSummary);
    System.out.println(salaireSummary.getCount());
    System.out.println(salaireSummary.getMin());
    System.out.println(salaireSummary.getMax());
    System.out.println(salaireSummary.getAverage());
    System.out.println(salaireSummary.getSum());

DoubleSummaryStatistics{count=6,sum=13480,000000, min=1280,000000, average=2246,666667,
 max=3300,000000}
6
1280.0
3300.0
2246.6666666666665
13480.0

Les méthodes averagingInt(), averagingLong() et averagingDouble() de la classe Collectors renvoient un Collector qui calcule la moyenne des données numériques extraites des éléments du Stream.

package fr.jmdoudoux.dej.streams;

import java.util.stream.Collectors;
import java.util.stream.Stream;

public class TestCollectors {

  public static void main(String[] args) {
    Stream<String> chaines = Stream.of("aaa", "bb", "ccccc");
    Double moyenne = chaines.collect(Collectors.averagingDouble(String::length));
    System.out.println(moyenne);
  }
}

3.3333333333333335

Les méthodes summingInt(), summingLong() et summingDouble() de la classe Collectors renvoient un Collector qui calcule la somme des données numériques extraites des éléments du Stream.

package fr.jmdoudoux.dej.streams;

import java.util.stream.Collectors;
import java.util.stream.Stream;

public class TestCollectors {

  public static void main(String[] args) {
    Stream<String> chaines = Stream.of("aaa", "bb", "ccccc");
    long somme = chaines.collect(Collectors.summingLong(String::length));
    System.out.println(somme);
  }
}

10

Les méthodes minBy() et maxBy() de la classe Collectors renvoient un Collector qui détermine respectivement le plus petit et le plus grand élément du Stream. Elles attendent en paramètre un Comparator qui sera utilisé pour déterminer l'élément concerné. Elles renvoient un Optional qui encapsule éventuellement cet élément s'il existe.

package fr.jmdoudoux.dej.streams;

import java.util.Comparator;
import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class TestCollectors {

  public static void main(String[] args) {
    Stream<String> chaines = Stream.of("aaa", "bb", "ccccc");
    Optional<String> lePlusGrand = chaines.collect(
      Collectors.maxBy(Comparator.naturalOrder()));
    System.out.println(lePlusGrand.get());
  }
}

ccccc

Le Comparator utilisé peut par exemple être plus générique en utilisant la méthode static comparing() de l'interface Comparator.

package fr.jmdoudoux.dej.streams;

import java.util.Comparator;
import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class TestCollectors {

  public static void main(String[] args) {
    Stream<String> chaines = Stream.of("aaa", "bb", "ccccc");
    Optional<String> lePlusLong = chaines.collect(
      Collectors.maxBy(Comparator.comparingInt(String::length)));
    System.out.println(lePlusLong.get());
  }
}

22.5.2.5. Les fabriques qui renvoient des Collectors pour effectuer des groupements

La méthode groupingBy() renvoie un Collector qui va regrouper les éléments du Stream dans une Map. Elle possède plusieurs surcharges :

La surcharge groupingBy(Function<? super T,? extends K> classifier) renvoie un Collector qui permet de grouper les éléments selon la clé obtenue par la fonction de classification fournie. Elle utilise implicitement le downstream Collector toList().

package fr.jmdoudoux.dej.streams;

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class TestCollectors {

  public static void main(String[] args) {
    Stream<String> chaines = Stream.of("aaa", "bb", "ccccc");
    Map<Integer, List<String>> map = chaines.collect(
      Collectors.groupingBy(String::length, Collectors.toList()));
    for (Map.Entry entry : map.entrySet()) {
      System.out.println(entry.getKey() + ", " + entry.getValue());
    }
  }
}

2, [bb]
3, [aaa]
5, [ccccc]

La méthode partitioningBy() est une spécialisation de la méthode groupingBy(). Elle attend en paramètre un Predicate pour grouper les éléments selon la valeur booléenne retournée par le Predicate. La collection de type Map retournée possède donc forcément un booléen comme clé.

package fr.jmdoudoux.dej.streams;

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class TestCollectors {

  public static void main(String[] args) {
    Stream<String> chaines = Stream.of("aaa", "bb", "ccccc");
    Map<Boolean, List<String>> map = chaines.collect(
      Collectors.partitioningBy(s -> s.length() >= 3));
    for (Map.Entry entry : map.entrySet()) {
      System.out.println(entry.getKey() + ", " + entry.getValue());
    }
  }
}

false, [bb]
true, [aaa, ccccc]

22.5.2.6. Les fabriques qui renvoient des Collectors pour effectuer des transformations

La méthode mapping() renvoie un Collector qui va transformer les objets de type T en type U grâce à la Function avant d'appliquer le downstream Collector.

public static <T,U,A,R> Collector<T,?,R> mapping(Function<? super T,? extends U> mapper, Collector<? super U,A,R> downstream)

   Map<String, Set<String>> nomsGroupesParVille
       = personnes.stream()
                  .collect(groupingBy(Personne::getVille,
                           mapping(Personne::getNom, toSet())));

La méthode reducing() renvoie un Collector qui permet de réaliser une opération de réduction appliquée de manière répétée sur tous les éléments du Stream pour produire un résultat.

Elle possède plusieurs surcharges :

Ces surcharges attendent jusqu'à trois paramètres :

• Identity : la valeur initiale qui sera retournée si le Stream est vide
• Mapper : une fonction à appliquer sur chacun des éléments pour extraire la valeur à utiliser
• Op : une opération qui combine les valeurs

       int tailleTotale = personnes.stream()
           .collect(reducing( 0, Personne::getTaille, Integer::sum));

On peut obtenir le même comportement pour certains de ces Collectors en utilisant des opérations de l'interface Stream notamment min(), max() ou reduce() qu'il est préférable d'utiliser. Cependant l'utilisation de ces Collectors est parfois requise notamment pour combiner des Collectors afin de réaliser des opérations plus complexes essentiellement en tant que downstream Collector.

22.5.3. La composition de Collectors

Il est possible de combiner des Collectors pour réaliser des réductions plus complexes : par exemple faire des groupements à plusieurs niveaux.

Ces combinaisons sont similaires à celles utilisables en SQL : il est possible de combiner un GROUP BY avec des opérations telles que COUNT. Il est donc possible d'utiliser un autre Collector pour déterminer la valeur comme par exemple pour compter le nombre d'éléments de chaque groupe.

Une surcharge de la méthode groupBy() qui attend en second paramètre un objet de type Collector permet d'appliquer le Collector pour réduire les éléments du groupe et ainsi obtenir la valeur associée à la clé.

    Stream<String> mots = Stream.of("aa", "bb", "aa", "bb","cc", "bb");
    Map<String, Long> nbMots = mots.collect(
      Collectors.groupingBy(s -> s.toUpperCase(), Collectors.counting()));
    for (Map.Entry entry : nbMots.entrySet()) {
      System.out.println(entry.getKey() + ", " + entry.getValue());
    }

CC, 1
BB, 3
AA, 2

Comme la méthode groupingBy() renvoie un Collector, il est possible d'utiliser un groupingBy() comme downstream Collector et ainsi réaliser un groupement à deux niveaux. Dans l'exemple ci-dessous, on effectue un groupement par âge et par salaire des employés.

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

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

public class TestStream {

  public static void main(String[] args) {
    employes.add(new Employe("n1", 33, 29000));
    employes.add(new Employe("n2", 41, 41000));
    employes.add(new Employe("n3", 33, 29000));
    employes.add(new Employe("n4", 41, 37000));
    employes.add(new Employe("n5", 33, 33000));
    employes.add(new Employe("n6", 54, 54000));
    
    Map<Integer, Map<Double, List<Employe>>> employesParAgeEtParSalaire = employes
        .stream()
        .collect(groupingBy(Employe::getAge,
                 groupingBy(Employe::getSalaire)));
    System.out.println("resultat = "+ employesParAgeEtParSalaire);
  }
}

resultat = {33={33000.0=[Employe [nom=n5]],
29000.0=[Employe [nom=n1], Employe [nom=n3]]}, 54={54000.0=[Employe [nom=n6]]},
41={37000.0=[Employe [nom=n4]], 41000.0=[Employe [nom=n2]]}}

La combinaison de Collectors au travers des downstream Collectors permet d'exprimer des traitements complexes. Dans l'exemple ci-dessous, ils sont utilisés pour déterminer le salaire le plus élevé par tranches d'âges d'une collection d'employés.

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

import static java.util.stream.Collectors.groupingBy;
import static java.util.stream.Collectors.mapping;
import static java.util.stream.Collectors.maxBy;
import java.util.ArrayList;
import java.util.List;
importjava.util.Map;

public class TestStream {

  public static void main(String[] args) {
    List<Employe> employes = new ArrayList<>(6);
    employes.add(new Employe("n1", 33, 29000));
    employes.add(new Employe("n2", 41, 41000));
    employes.add(new Employe("n3", 33, 29000));
    employes.add(new Employe("n4", 41, 37000));
    employes.add(new Employe("n5", 33, 33000));
    employes.add(new Employe("n6", 54, 54000));

    Map<Integer, Double> salaireMaxParAges = employes.stream()
        .collect(groupingBy(Employe::getAge,
            mapping(Employe::getSalaire,
               collectingAndThen(maxBy(Double::compare),
                    d -> d.get()
                        .doubleValue()))));
    System.out.println(salaireMaxParAges);
  }
}

{33=33000.0, 54=54000.0, 41=41000.0}

22.5.4. L'implémentation d'un Collector

Le JDK propose en standard un ensemble d'implémentations de Collector pour des besoins courants. Il est possible de développer sa propre implémentation de l'interface Collector pour définir des opérations de réductions personnalisées si celles-ci ne sont pas fournies par le JDK.

L'interface Collector implémentée doit être typée avec 3 génériques :

public interface Collector<T, A, R> {...}

Les trois types génériques correspondent aux types utilisés par le Collector :

• T : le type des objets qui seront traités
• A : le type de l'objet utilisé comme accumulator
• R : le type de l'objet qui sera retourné comme résultat

Il est courant que les types A et R soient identiques : c'est par exemple le cas pour un type de l'API Collection. Mais ils peuvent être différents par exemple un accumulator de type StringBuilder et un résultat de type String.

L'implémentation d'un Collector peut se faire de deux manières :

• Utiliser une des surcharges de la fabrique of() de l'interface Collector : elle permet de créer des Collector simples
• Définir une classe qui implémente l'interface Collector : pour des cas plus complexes ou des besoins particuliers

22.5.5. Les fabriques of() pour créer des instances de Collector

Il est possible de créer une instance de type Collector en utilisant sa fabrique of() qui possède deux surcharges.

L'exemple ci-dessous créé une instance de type Collector pour concaténer les prénoms d'un Stream de d'objet de type Personne mis en majuscules. Le séparateur utilisé est une virgule.

  Collector<Personne, StringJoiner, String> prenomPersonneCollector =
      Collector.of(
       () -> new StringJoiner(","),                       // supplier

       (sj, p) -> sj.add(p.getPrenom().toUpperCase()),  // accumulator

       (sj1, sj2) -> sj1.merge(sj2),                    // combiner

       StringJoiner::toString);                        // finisher

  String prenoms = personnes
   .stream()
   .collect(prenomPersonneCollector);
  System.out.println(prenomss);

Le Collector utilise la classe StringJoiner de Java 8 pour concaténer les prénoms des personnes dans ses traitements :

• Le supplier fournit une instance de StringJoiner avec le délimiteur à utiliser
• L'accumulator utilise la méthode add() du StringJoiner pour concaténer chaque prénom mis en majuscules en les séparant par une virgule
• Le combiner utilise la méthode merge() pour fusionner les deux StringJoiner fournis en paramètre
• Le finisher utilise la méthode toString() du StringJoiner pour fournir le résultat final sous la forme d'une chaîne de caractères. Son utilisation est nécessaire car le type du Supplier et le type renvoyé du Collector sont différents

22.6. Les Streams pour des données primitives

L'interface Stream<T> utilise un type generic T et s'utilise donc avec des objets de type T. Il existe aussi plusieurs interfaces dédiées à la manipulation de types primitifs int, long et double respectivement IntStream, LongStream et DoubleStream.

Lorsqu'un Stream est utilisé sur des données primitives cela peut engendrer de nombreuses opérations de boxing/unboxing pour encapsuler une valeur primitive dans un objet de type wrapper et vice versa. Ces opérations peuvent être coûteuses lorsque le nombre d'éléments à traiter dans le Stream est important.

L'exemple ci-dessus est purement pédagogique pour réaliser des traitements sur un Stream qui effectue un ensemble d'opération sur des valeurs entières.

public Long testStreamLong() {
  Long somme = Stream
      .iterate(0L, i -> i + 1L)
      .limit(20_000_000)
      .filter(i -> (i % 2) == 0)
      .map(i -> i + 1)
      .sorted()
      .reduce(0L, Long::sum);
  return somme;
}

Un benchmark est réalisé sur cette méthode avec JMH :

Benchmark                       Mode  Cnt   Score    Error     Units
StreamBenchmark.testStreamLong  avgt  10    4322,430 ± 258,794 ms/op

La méthode ci-dessous produit le même résultat mais elle utilise un Stream primitif

public long testLongStream() {
  long somme = LongStream
      .iterate(0, i -> i + 1)
      .limit(20_000_000)
      .filter(i -> (i % 2) == 0)
      .map(i -> i + 1)
      .sorted()
      .sum();
  return somme;
}

Un benchmark est réalisé sur cette méthode pour permettre de comparer les performances des deux versions :

Benchmark                       Mode Cnt    Score   Error   Units
StreamBenchmark.testLongStream  avgt 10   233,061 ± 10,962  ms/op
StreamBenchmark.testStreamLong  avgt 10  4322,430 ± 258,794 ms/op

La différence de temps de traitement est sans appel en faveur de l'utilisation du Stream primitif. Lorsque les données à traiter par le Stream sont des données primitives de type int, long ou double, il est donc préférable d'utiliser les interfaces IntStream, LongStream et DoubleStream.

C'est par exemple un IntStream qui est retourné lorsque l'on utilise une opération de type map T vers int. Dans ce cas, ce sont des interfaces fonctionnelles renvoyant un type primitif qui sont utilisées comme IntSupplier, IntConsumer, ToIntFunction.

Ces interfaces fonctionnent comme l'interface Stream avec quelques différences :

• Les interfaces fonctionnelles utilisées sont celles spécifiques aux types primitifs : par exemple IntFunction à la place de Function, IntPredicate à la place de Predicate, ...
• Elles proposent quelques opérations terminales spécifiques aux types primitifs comme sum(), average(), ...

package fr.jmdoudoux.dej.streams;

import java.util.stream.IntStream;

public class TestStream {

  public static void main(String[] args) { 
    int valeur = IntStream.of(1, 2, 3, 4, 5).map(n -> 2 * n).sum();
    System.out.println(valeur);
  }
}

30

Il est parfois utile voire même nécessaire de transformer un Stream primitif en Stream d'objets et vice versa.

Les méthodes mapToInt(), mapToLong() et mapToDouble() de l'interface Stream fonctionne comme la méthode map() sauf qu'au lieu de renvoyer un Stream<T>, elles renvoient respectivement un IntStream, un LongStream et un DoubleStream.

Elles permettent donc de transformer un Stream d'objets en Stream de primitifs de leur type respectif. Elles attendent respectivement en paramètre une interface fonctionnelle de type ToIntFunction, ToLongFunction et ToDoubleFunction.

package fr.jmdoudoux.dej.streams;

import java.util.stream.Stream;

public class TestStream {

  public static void main(String[] args) {
    Stream.of("1", "2", "3")
          .mapToInt(Integer::parseInt)
          .max().ifPresent(System.out::println);
  }
}

3

package fr.jmdoudoux.dej.streams;

import java.util.stream.Stream;

public class TestStream {

  public static void main(String[] args) {
    int valeur = Stream.of(1.0, 2.0, 3.0).mapToInt(Double::intValue).sum();
    System.out.println(valeur);
  }
}

6

Les Streams primitifs peuvent être transformés en Stream d'objets en utilisant la méthode mapToObj(). Selon le type primitif, elle attend respectivement en paramètre une interface fonctionnelle de type IntFunction, LongFunction et DoubleFunction.

package fr.jmdoudoux.dej.streams;

import java.util.stream.IntStream;

public class TestStream {

  public static void main(String[] args) {
    IntStream.range(1, 4).mapToObj(i -> "" + i).forEach(System.out::println);
  }
}

1
2
3

La méthode boxed() permet de transformer un Stream primitif en Stream<T> ou T est soit un Integer, un Long ou un Double.

22.6.1. Les interfaces IntStream, LongStream et DoubleStream

L'interface IntStream propose de nombreuses méthodes :

Les méthodes range() et rangeClosed() permettent de créer un Stream dont les éléments seront les valeurs comprises entre les bornes fournies en paramètres :

• Les deux méthodes attendent en premier paramètre la valeur initiale de la plage de valeur
• Le second paramètre indique la valeur finale, exclue pour la méthode range() et incluse pour la méthode rangeClosed()

package fr.jmdoudoux.dej.streams;

import java.util.stream.IntStream;

public class TestStream {

  public static void main(String[] args) {
    IntStream.rangeClosed(1, 5)
             .forEach(n -> System.out.print(n));
    System.out.println();
    IntStream.range(1, 5)
             .forEach(n -> System.out.print(n));
  }
}

12345
1234

L'exemple ci-dessous détermine les nombres impairs compris entre 0 et 10 inclus.

    IntStream nombresImpairs = IntStream.rangeClosed(0, 10)
                                        .filter(nombre -> (nombre % 2) == 1);
    nombresImpairs.forEach(System.out::println);

1
3
5
7
9

Il est possible de créer une collection d'objets en utilisant la méthode range() pour définir chaque élément.

    List<Departement> departements = new ArrayList<>();
    IntStream.range(1, 3).forEach(i -> {
      Departement departement = new Departement("Departement" + i);
      departements.add(departement);
      IntStream.range(1, 4).forEach(
          j -> departement.getEtudiants().add(new Etudiant("nom" + j + "_"
            + departement.getNom(), 20 + j)));
    });

Remarque : l'utilisation de Streams dont les traitements modifient l'état d'objets externes n'est cependant pas recommandé.

La méthode sum() est une opération de réduction qui calcule la somme des éléments du Stream.

   int tailleTotale = personnes.stream()
                               .mapToInt(Personne::getTaille)
                               .sum();
    System.out.println(tailleTotale);

La méthode average() est une opération de réduction qui calcule la moyenne des éléments du Stream. Elle renvoie un objet de type OptionalDouble. Elle renvoie un objet de type xxxOptional ou xxx est le type primitif du Stream.

package fr.jmdoudoux.dej.streams;

import java.util.stream.IntStream;

public class TestStream {

  public static void main(String[] args) {
    IntStream.of(1, 2, 3, 4, 5)
             .map(n -> 2 * n)
             .average()
             .ifPresent(System.out::println);
  }
}

6.0

22.7. L'utilisation des Streams avec les opérations I/O

Certaines méthodes des classes I/O renvoient un Stream notamment à partir de la lecture d'un fichier texte ou le contenu des éléments d'un répertoire.

22.7.0.1. La création d'un Stream à partir d'un fichier texte

L'API NIO 2 de Java 7, propose une méthode pour lire l'intégralité des lignes d'un fichier texte.

List<String> lines = Files.readAllLines(somePath, someCharset);

L'utilisation d'un Stream permet de ne pas avoir besoin de stocker l'intégralité du contenu du fichier en mémoire : la lecture dans le fichier se fait au fur et à mesure de la consommation par le Stream. Ceci peut être intéressant notamment pour le traitement de gros fichiers.

La méthode lines() de la classe Files attend en paramètre le fichier sous la forme d'un objet de type Path. Elle permet de renvoyer un Stream<String> qui va lire de manière lazy les lignes d'un fichier pour alimenter le Stream au fur et à mesure de la consommation des lignes.

Contrairement à la méthode readAllLines() qui lit l'intégralité du fichier, la méthode lines() peut se contenter de ne lire que les lignes que lors de l'invocation de la méthode terminale et la lecture peut être interrompue par l'exécution d'une méthode de type short-circuiting.

Les caractères sont décodés en utilisant le ChartSet UTF-8 par défaut. La méthode lines() possède une seconde surcharge qui attend en paramètre deux objets de type Path et CharSet.

Lorsque la source de données du Stream effectue des opérations de type I/O, il est nécessaire de libérer les ressources allouées pour lire les données. L'interface BaseStream définit la méthode close() et la classe Stream implémente l'interface AutoCloseable.

La plupart des Streams n'ont pas besoin d'avoir leur méthode close() invoquée sauf dans certains cas notamment lorsque la source réalise des opérations de type I/O. La méthode close() doit être explicitement invoquée si nécessaire pour libérer les ressources ouvertes par la source.

Pour éviter de laisser le système libérer les ressources au bout d'un certain timeout, il faut invoquer la méthode close() de l'objet responsable de la lecture des données. Ceci peut éviter des fuites de ressources. Il est donc important d'invoquer la méthode close() du Stream pour qu'elle invoque la méthode close() de la source utilisée pour lire les lignes du fichier.

La méthode onClose() de l'interface BaseStream est une opération intermédiaire qui permet d'exécuter un traitement lorsque la méthode close() du Stream est invoquée. Il est possible d'invoquer plusieurs fois cette méthode : les traitements seront alors exécutés dans leur ordre d'ajout.

   try {
     Stream<String> lignes = Files.lines(Paths.get("fichier.txt"), Charset.defaultCharset());
     long nbLignes = lignes.count();
     lignes.close();
     System.out.println("Nb lignes = " + nbLignes);
   } catch (IOException e) {
      e.printStackTrace();
   }

Il est aussi possible d'utiliser un try with resources sur un Stream qui va invoquer sa méthode close(). Cette dernière va se charger de libérer les éventuelles ressources ouvertes par la source de données.

    try (Stream<String> lignes = Files.lines(Paths.get("fichier.txt"))) {
      long nbLignes = lignes.count();
      System.out.println("Nb lignes = " + nbLignes);
    } catch (IOException e) {
      e.printStackTrace();
    }

La méthode lines() de la classe BufferedReader renvoie un Stream dont les éléments sont les lignes du fichier lu.

Il est important que seul le Stream utilise le BufferedReader pour lire les données.

Si une exception de type IOException survient durant l'utilisation du BufferedReader, celle-ci est encapsulée dans une UncheckedIOException.

    try {
      BufferedReader br = Files.newBufferedReader(Paths.get("fichier.txt"));
      Stream<String> lignes = br.lines();
      lignes.forEach(System.out::println);
      lignes.close();
    } catch (Exception e) {
      e.printStackTrace();
    }

Il est aussi plus simple d'utiliser le Stream comme ressource d'un try with resources.

    try (Stream<String> lignes = Files.newBufferedReader(Paths.get("fichier.txt"))
         .lines()
         .onClose(() -> System.out.println("Fermeture du fichier"))) {
      lignes.forEach(System.out::println);
    } catch (Exception e) {
      e.printStackTrace();
    }

22.7.0.2. La création d'un Stream à partir du contenu d'un répertoire

Plusieurs méthodes ont été ajoutées à la classe Files dans la version 8 de Java.

Il est important d'invoquer la méthode close() du Stream à la fin de son utilisation. Le plus simple est d'utiliser une instruction try with resources.

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.stream.Stream;

public class TestFilesStream {

  public static void main(String[] args) {
    String folder = "c:/temp";
    try (Stream<Path> paths = Files.list(Paths.get(folder))) {
      paths.filter(p -> p.toString()
           .endsWith(".dat"))
           .forEach(System.out::println);
    } catch (IOException ioe) {
      ioe.printStackTrace();
    }
  }
}

22.8. Le traitement des opérations en parallèle

Une des fonctionnalités les plus mises en avant concernant les Streams est la facilité déconcertante pour exécuter les traitements des opérations en parallèle.

L'avènement des processeurs multi-cour, omniprésents dans les appareils informatique (ordinateurs, tablettes, téléphones mobiles, ...) oblige à mettre en oeuvre des traitements en parallèle pour exploiter leur puissance.

Java 7 a introduit le framework Fork/Join pour faciliter la parallélisation de tâches. Ce framework est utilisé par l'implémentation de l'API Stream pour permettre l'exécution de ses traitements en parallèle.

Globalement aussi, le volume des données à traiter augmente. Généralement ces données sont stockées dans une collection. La conception de l'API Collection étant assez ancienne (Java 1.2), il n'est pas facile de proposer l'intégration de fonctionnalités en parallèle dans ces classes.

L'exécution de traitements sur des données d'une collection requiert de réaliser une itération sur chacun des éléments. Le traitement d'éléments dans une itération est par définition intrinsèquement séquentiel.

List<String> elements = Arrays.asList("elem1", " elem2", " elem3","elem4", "elem5");
for (String element : elements)
   System.out.println(element);

La mise en oeuvre de ces traitements en parallèle est très compliquée même avec l'API Fork/Join. Pourtant elle peut être nécessaire si le volume des données est important. Pour ne pas réécrire intégralement l'API Collection et ainsi maintenir la rétrocompatibilité, Java SE 8 propose dans l'API Streams le traitement en parallèle de données.

Très facilement, l'API Stream permet de réaliser ses traitements en séquentiel ou en parallèle. Seules des méthodes par défaut pour l'obtention d'un Stream à partir d'une Collection ont été ajoutées à cette API.

List<String> elements = Arrays.asList("elem1", " elem2", " elem3", "elem4", "elem5");
elements.stream()
        .forEach(System.out::println);
elements.parallelStream()
        .forEach(System.out::println);

22.8.1. La mise en oeuvre des Streams parallèles

La plupart des Streams créés dans l'API du JDK ont leurs opérations qui par défaut seront exécutées de manière séquentielle dans le thread courant. Il est alors nécessaire de préciser explicitement que l'on souhaite que les opérations du Stream soient exécutées de manière parallèle.

Puisque c'est l'API qui se charge d'itérer sur les différents éléments pour exécuter le pipeline d'opérations, il est facilement possible de demander l'exécution de ces traitements en parallèle. Le fonctionnement interne de l'API masque alors toute la complexité de l'exécution en parallèle des traitements.

Pour le développeur, cela consiste simplement :

• Soit à demander la création d'un Stream parallèle en utilisant la méthode parallelStream() des collections
• Soit à convertir un Stream séquentiel en Stream parallèle invoquant la méthode parallel() de l'interface BaseStream qui est une opération intermédiaire

Remarque : il est aussi possible de transformer un Stream parallèle en Stream séquentiel en invoquant la méthode sequential().

22.8.2. Le fonctionnement interne d'une Stream

Le découpage en sous-lots est réalisé grâce à un objet de type Spliterator. La suite d'opérations du pipeline pour chacun de ces sous-lots sera alors exécuté par un thread libre du pool du framework Fork/Join.

22.8.2.1. L'interface Spliterator pour l'obtention des données par la source

La liaison entre un Stream et sa source de données se fait avec un objet de type Spliterator : il permet l'obtention des données de la source par le Stream.

Un Spliterator est un objet pour traverser et partitionner les éléments d'une source. Il offre donc deux fonctionnalités :

• Un itérateur qui permet de parcourir séquentiellement les éléments d'une source et de les fournir au Stream
• La décomposition de l'ensemble des éléments en deux sous-ensembles pour leur traitement en parallèle par le Stream, idéalement contenant chacun une moitié, dans le Spliterator courant et dans un autre Spliterator. Ces deux Spliterator peuvent être eux-mêmes ensuite décomposés

L'interface Spliterator définit les méthodes pour un objet qui permet de parcourir et partitionner les éléments d'une source de données :

Le parcours des éléments peut se faire avec deux méthodes :

• La méthode tryAdvance() permet d'appliquer le Consumer aux éléments un par un (de manière similaire à l'invocation de la méthode next() de l'interface Iterator)
• La méthode forEachRemaining() permet d'appliquer le Consumer à tous les éléments restants

L'interface Spliterator propose le support de la création de lots d'éléments traités en parallèle dans plusieurs threads par l'API Stream. Un Spliterator permet de découper un ensemble plus petit des éléments d'entrée dans un nouveau Spliterator (idéalement, la moitié), et laisser le reste des éléments dans le Spliterator original. Les Spliterator peuvent ensuite être de nouveau décomposés au besoin. La méthode trySplit() renvoie une nouvelle instance de type Spliterator dont la responsabilité est de gérer un sous-ensemble des éléments du Spliterator. La méthode trySplit() peut renvoyer null si l'implémentation du Spliterator ne supporte pas la création de lots : attention dans ce cas, le Stream utilisant un tel Spliterator ne pourra pas être exécuté en parallèle. L'implémentation du Spliterator utilisée peut affecter les performances d'un Stream exécuté en parallèle.

Un Spliterator contient des métadonnées comme par exemple le nombre d'éléments à traiter (s'il est connu) et un ensemble de quelques caractéristiques sous la forme d'un entier. Ces informations peuvent être utilisées pour optimiser les traitements réalisés par le Spliterator.

L'ensemble de caractéristiques (distinct, ordered, sorted, sized, nonnull, immutable, concurrent, subsized) dépend du type de source de données et est utilisé pour réaliser ses traitements de manière plus ou moins optimisée.

Pour des usages standards, l'implémentation de Spliterator utilisée est celle fournie par la méthode qui permet d'obtenir un Stream. Pour des besoins particuliers, il peut être nécessaire d'implémenter son propre Spliterator. Il suffit de définir une classe qui implémente l'interface Spliterator.

Plusieurs caractéristiques doivent être prise en compte lors du développement d'un Spliterator :

• Le Spliterator connait le nombre d'éléments de la source de données
• Le Spliterator est capable de partitionner les éléments de la source de données
• Le Spliterator peut partitionner les éléments en portion de taille similaire (caractéristique SUBSIZED)

La manière la plus simple d'obtenir un Spliterator mais avec des performances moyennes voire mauvaises est d'utiliser la méthode spliteratorUnknownSize() de la classe Spliterators en lui passant en paramètre un Iterator.

Pour obtenir de meilleure performance, il est possible d'utiliser la méthode spliterator() de la classe Spliterators en lui passant en paramètre un Iterator et le nombre d'éléments.

Il est enfin possible de définir son propre Spliterator en implémentant l'interface Spliterator.

Les classes de l'API Collection possèdent des implémentations de Spliterator : il est possible de s'en inspirer pour développer ses propres implémentations. Les interfaces Iterable et Collection proposent une implémentation basique et peu optimisée du Spliterator retournée par la méthode spliterator(). Les interfaces filles et leurs implémentations proposent des implémentations beaucoup plus performantes de leurs Spliterators notamment grâce à une exploitation de leurs métadonnées. Par exemple, l'implémentation proposée par ArrayList utilise le nombre d'éléments contenus dans la collection pour déterminer de manière efficace le nombre d'éléments de chaque lot retourné par la méthode split().

Les implémentations de Spliterator doivent faire un compromis entre simplicité d'implémentation et performance d'exécution.

Une implémentation basique mais peu performante est retournée par la méthode spliteratorUnkownSize(Iterator, int) de la classe Spliterators : celle-ci n'utilise pas le nombre d'éléments puisqu'il n'est pas connu par un Iterator. L'implémentation est donc obligée de parcourir les éléments de l'Iterator et utilise un algorithme de découpage rudimentaire.

Une implémentation plus performante pourra par exemple s'appuyer sur le nombre d'éléments de la source de données si celui-ci est connu. L'implémentation pourra alors utiliser cette information pour définir des lots de données équilibrés et de manière efficace surtout si par exemple, les données sont stockées dans un tableau. L'implémentation d'un Spliterator peut aussi utiliser d'autres caractéristiques pour optimiser ses traitements.

Plusieurs classes du JDK ont aussi été modifiées pour servir de source de données pour un Stream. Il est aussi possible d'adapter une classe qui encapsule des données pour qu'elle puisse être utilisée comme source pour un Stream. Pour créer un Stream à partir d'une classe qui sera sa source de données, il est nécessaire d'utiliser un Spliterator.

La classe StreamSupport est un utilitaire qui facilite la création de Streams. Elle propose plusieurs méthodes statiques pour créer un Stream qui attendent en paramètre un Spliterator et un booléen qui précise si le Stream est séquentiel (false) ou parallèle (true).

22.8.2.2. L'utilisation du framework Fork/Join

Le code utilisé pour paralléliser un traitement et le code de ce traitement exécuté de manière séquentielle est très différent : le premier nécessite de nombreuses lignes de code supplémentaires.

Le framework Fork/Join propose une API pour faciliter l'implémentation de traitements en parallèle sur un ensemble de données.

Le mode de fonctionnement de ce framework est composé de plusieurs étapes :

• Découper l'ensemble des données en sous-ensemble
• Traiter chaque sous-ensemble dans un des threads du pool pour produire un résultat partiel
• Agréger les résultats partiels pour produire un résultat final

L'API Stream propose d'utiliser le même code pour définir les traitements. Ceux-ci pourront très facilement être exécutés de manière séquentielle ou parallèle. Ceci rend très facile d'utiliser l'exécution des traitements d'un Stream en parallèle de manière fiable puisque c'est l'implémentation de l'API Stream qui se charge de tout en utilisant le framework Fork/Join.

Le framework Fork/Join utilise un pool de threads de type ForkJoinPool. A partir de Java 8, Fork/Join utilise un pool de threads commun. Il est possible d'obtenir l'instance de ce pool en invoquant la méthode statique commonPool() de la classe ForkFoinPool.

La méthode getParallelism() permet d'obtenir le nombre de threads souhaité dans le pool et la méthode getCommonParallelism() permet d'obtenir le nombre de threads souhaité dans le pool commun.

Par défaut, le nombre de threads souhaité dans le pool pour l'exécution des tâches par le framework Fork/Join est égale au nombre de coeurs moins un. Par exemple, sur une machine équipée d'un processeur Intel Core i7-5500U, la valeur de l'attribut parallelism est 3.

package fr.jmdoudoux.dej.streams;

import java.util.concurrent.ForkJoinPool;

public class TestParellelStream {

  public static void main(String[] args) {
    ForkJoinPool commonPool = ForkJoinPool.commonPool();
    System.out.println(commonPool.getParallelism());
  }
}

3

Il est possible de modifier cette valeur en utilisant la variable d'environnement de la JVM java.util.concurrent.ForkJoinPool.common.parallelism

-Djava.util.concurrent.ForkJoinPool.common.parallelism=6

Pour comprendre le fonctionnement de l'exécution d'un Stream parallèle, il est possible d'afficher des informations lors de l'exécution de chaque opération : la méthode invoquée, l'élément traité et le nom du thread courant.

package fr.jmdoudoux.dej.streams;

import java.io.PrintStream;
import java.util.stream.IntStream;

public class TestParellelStream {

  public static void main(String[] args) {
    IntStream
        .rangeClosed(1, 5)
        .parallel()
        .mapToObj(v -> {
          afficher("mapToObj", "e" + v);
          return "e" + v;
        })
        .filter(s -> {
          afficher("filter ", s);
          return true;
        })
        .map(s -> {
          afficher("map ", s);
          return s.toUpperCase();
        })
        .forEach(s -> afficher("forEach ", s));
  }
  
  private static PrintStream afficher(String methode, String valeur) {
    return System.out.format("%s: %s [%s]\n", methode, valeur, 
      Thread.currentThread().getName());
  }
}

mapToObj: e3 [main]
mapToObj: e1 [ForkJoinPool.commonPool-worker-3]
mapToObj: e5 [ForkJoinPool.commonPool-worker-2]
mapToObj: e2 [ForkJoinPool.commonPool-worker-1]
filter: e5 [ForkJoinPool.commonPool-worker-2]
filter: e1 [ForkJoinPool.commonPool-worker-3]
filter: e3 [main]
map: e3 [main]
map: e1 [ForkJoinPool.commonPool-worker-3]
forEach: E1 [ForkJoinPool.commonPool-worker-3]
map: e5 [ForkJoinPool.commonPool-worker-2]
filter: e2 [ForkJoinPool.commonPool-worker-1]
map: e2 [ForkJoinPool.commonPool-worker-1]
forEach: E5 [ForkJoinPool.commonPool-worker-2]
mapToObj:e4 [ForkJoinPool.commonPool-worker-3]
forEach: E3 [main]
filter: e4 [ForkJoinPool.commonPool-worker-3]
map: e4 [ForkJoinPool.commonPool-worker-3]
forEach: E2 [ForkJoinPool.commonPool-worker-1]
forEach: E4 [ForkJoinPool.commonPool-worker-3]

L'exécution des traitements en parallèle n'est pas prédictive. Plusieurs exécutions de ce code donneront un ordre de traces différent.

L'API Stream utilise le thread courant et les threads du pool commun de Fork/Join. Chaque thread exécute le pipeline d'opérations pour un sous-ensemble de données, restreint dans l'exemple ci-dessus car le nombre d'éléments est trop faible.

22.8.3. Le mode d'exécution des Streams

Lorsque l'opération terminale du Stream est invoquée, les opérations sont exécutées en séquentiel ou en parallèle selon la configuration du Stream.

Les opérations parallel() et sequential() de l'interface BaseStream  sont des opérations intermédiaires : elles sont donc lazy.

package fr.jmdoudoux.dej.streams;

import java.io.PrintStream;
import java.util.stream.IntStream;

public class TestParellelStream {

  public static void main(String[] args) {
    IntStream
        .rangeClosed(1, 5)
        .mapToObj(v -> {
          afficher("mapToObj", "e" + v);
          return "e" + v;
        })
        .filter(s -> {
          afficher("filter ", s);
          return true;
        })
        .map(s -> {
          afficher("map ", s);
          return s.toUpperCase();
        })
        .parallel()
        .forEach(s -> afficher("forEach ", s));
  }

  private static PrintStream afficher(String methode, String valeur) {
    return System.out.format("%s: %s [%s]\n", methode, valeur, 
      Thread.currentThread().getName());
  }
}

mapToObj:e2 [ForkJoinPool.commonPool-worker-1]
mapToObj:e3 [main]
mapToObj:e1 [ForkJoinPool.commonPool-worker-3]
mapToObj:e5 [ForkJoinPool.commonPool-worker-2]
filter: e1 [ForkJoinPool.commonPool-worker-3]
filter: e3 [main]
map: e3 [main]
filter: e2 [ForkJoinPool.commonPool-worker-1]
map: e2 [ForkJoinPool.commonPool-worker-1]
forEach: E3 [main]
map: e1 [ForkJoinPool.commonPool-worker-3]
filter: e5 [ForkJoinPool.commonPool-worker-2]
forEach: E1 [ForkJoinPool.commonPool-worker-3]
mapToObj: e4 [main]
forEach: E2 [ForkJoinPool.commonPool-worker-1]
filter: e4 [main]
map: e4 [main]
map: e5 [ForkJoinPool.commonPool-worker-2]
forEach: E4 [main]
forEach: E5 [ForkJoinPool.commonPool-worker-2]

C'est la dernière de ces opérations qui sera utilisée pour déterminer le mode d'exécution du Stream.

package fr.jmdoudoux.dej.streams;

import java.io.PrintStream;
import java.util.stream.IntStream;

public class TestParellelStream {

  public static void main(String[] args) {
    IntStream
        .rangeClosed(1, 5)
        .mapToObj(v -> {
          afficher("mapToObj", "e" + v);
          return "e" + v;
        })
        .parallel()
        .filter(s -> {
          afficher("filter ", s);
          return true;
        })
        .map(s -> {
          afficher("map ", s);
          return s.toUpperCase();
        })
        .sequential()
        .forEach(s -> afficher("forEach", s));
  }

  private static PrintStream afficher(String methode, String valeur) {
    return System.out.format("%s: %s [%s]\n", methode, valeur, 
      Thread.currentThread().getName());
  }
}

mapToObj: e1 [main]
filter: e1 [main]
map: e1 [main]
forEach: E1 [main]
mapToObj:e2 [main]
filter: e2 [main]
map: e2 [main]
forEach: E2 [main]
mapToObj:e3 [main]
filter: e3 [main]
map: e3 [main]
forEach: E3 [main]
mapToObj: e4 [main]
filter: e4 [main]
map: e4 [main]
forEach: E4 [main]
mapToObj: e5 [main]
filter: e5 [main]
map: e5 [main]
forEach: E5 [main]

De la même manière, bien que l'opération parallel() ait été utilisée puis la méthode sequential(), le Stream est exécuté de manière séquentielle dans son intégralité. Aucune opération n'est exécutée en parallèle.

22.8.4. Le comportement de certaines opérations en parallèle

L'exécution en séquentiel ou en parallèle d'un Stream permet généralement d'obtenir le même résultat sauf si une opération non déterministe est utilisée comme findAny(), findFirst(), ... Dans ce cas, le résultat n'est pas forcement le même sauf si le Stream ne contient plus qu'un seul élément lors de l'exécution de l'opération terminale.

Le comportement de l'opération sorted() est particulier lors d'une exécution en parallèle.

package fr.jmdoudoux.dej.streams;

import java.io.PrintStream;
import java.util.stream.IntStream;

public class TestParellelStream {

  public static void main(String[] args) {
    IntStream
        .rangeClosed(1, 5)
        .parallel()
        .mapToObj(v -> {
          afficher("mapToObj", "e" + v);
          return "e" + v;
        })
        .sorted((s1, s2) -> {
          afficher("sorted ", s1 + "/" + s2);
          return s1.compareTo(s2);
        })
        .forEach(s -> afficher("forEach ", s));
  }

  private static PrintStream afficher(String methode, String valeur) {
    return System.out.format("%s: %s [%s]\n", methode, valeur, 
      Thread.currentThread().getName());
  }
}

mapToObj: e2 [ForkJoinPool.commonPool-worker-1]
mapToObj: e5 [main]
mapToObj: e1 [ForkJoinPool.commonPool-worker-3]
mapToObj: e3 [ForkJoinPool.commonPool-worker-2]
mapToObj: e4 [ForkJoinPool.commonPool-worker-1]
sorted: e2/e1 [main]
sorted: e3/e2 [main]
sorted: e4/e3 [main]
sorted: e5/e4 [main]
forEach: e3 [main]
forEach: e5 [ForkJoinPool.commonPool-worker-2]
forEach: e2 [ForkJoinPool.commonPool-worker-1]
forEach: e1 [ForkJoinPool.commonPool-worker-3]
forEach: e4 [main]

L'exécution de la méthode sorted() pour chaque élément se fait toujours dans le même thread ce qui n'est pas forcement le comportement attendu vu que le Stream est parallèle.

En fait, l'implémentation de la méthode sorted() utilise la méthode parallelSort() de la classe Arrays. Les traitements de cette méthode sont exécutés de manière séquentielle jusqu'à un certain nombre d'éléments et une fois ce nombre atteint les traitements sont effectués en parallèle.

package fr.jmdoudoux.dej.streams;

import java.io.PrintStream;
import java.util.stream.IntStream;

public class TestParellelStream {
  public static void main(String[] args) {
    // ForkJoinPool commonPool = ForkJoinPool.commonPool();

    // System.out.println(commonPool.getParallelism());

    IntStream
        .rangeClosed(1, 10000)
        .parallel()
        .mapToObj(v -> {
          // afficher("mapToObj", "e" + v);

          return "e" + v;
        })
        .sorted((s1, s2) -> {
          afficher("sorted ", s1 + "/" + s2);
          return s1.compareTo(s2);
        })
        .forEach(s -> afficher("forEach ", s));
  }

  private static PrintStream afficher(String methode, String valeur) {
    return System.out.format("%s: %s [%s]\n", methode, valeur, 
      Thread.currentThread().getName());
  }
}

...
sorted: e9906/e9905 [ForkJoinPool.commonPool-worker-2]
sorted: e9907/e9906 [ForkJoinPool.commonPool-worker-2]
sorted: e6206/e6205 [ForkJoinPool.commonPool-worker-1]
sorted: e6207/e6206 [ForkJoinPool.commonPool-worker-1]
sorted: e6208/e6207 [ForkJoinPool.commonPool-worker-1]
sorted: e9908/e9907 [ForkJoinPool.commonPool-worker-2]
sorted: e9909/e9908 [ForkJoinPool.commonPool-worker-2]
sorted: e9910/e9909 [ForkJoinPool.commonPool-worker-2]
sorted: e9911/e9910 [ForkJoinPool.commonPool-worker-2]
sorted: e9912/e9911 [ForkJoinPool.commonPool-worker-2]
...

Si le nombre d'éléments à traiter est important, alors l'exécution de la méthode sorted() ce fait bien en parallèle.

Certaines opérations, comme reduce() ou collect() ont besoin de traitements supplémentaires lors de l'exécution du Stream en parallèle par rapport à leur exécution en séquentiel notamment pour agréger les résultats issus de chacun des threads.

22.8.5. L'impact de l'état ordonné ou non des éléments

Certains Stream peuvent avoir un ordre précis dans les éléments qu'il doit traiter. Dans ce cas, le Stream possède un attribut Ordered dont l'origine peut provenir soit :

• De la source de données qui peut déjà contenir les éléments dans un certain ordre (par exemple les tableaux, les collections de type List, ...) ou pas (par exemple les collections de type HashSet)
• Des opérations intermédiaires peuvent ordonner les éléments du Stream (par exemple l'opération sorted()) ou au contraire retirer le caractère ordonné des éléments (par exemple l'opération unordered())

Certaines opérations intermédiaires peuvent imposer un ordre sur les éléments d'un Stream qui n'est pas ordonné, alors que d'autres opérations intermédiaires peuvent retourner un Stream non ordonné pour un Stream ordonné.

Certaines opérations terminales peuvent ne pas tenir compte de l'ordre des éléments ordonnés d'un Stream (par exemple l'opération forEach() dans le cas d'un traitement en parallèle).

Pour un Stream ordonné, la plupart des opérations sont contraints de respecter l'ordre des éléments.

Dans un Stream séquentiel, si les éléments d'un Stream sont ordonnés, généralement la plupart des opérations intermédiaires traitent les éléments dans leur ordre. Pour un Stream séquentiel, le fait que les éléments soient ordonnés où non n'affecte pas les performances. Seul le déterminisme du résultat est impacté : si les éléments sont ordonnés, plusieurs exécutions donneront toujours le même résultat. Si les éléments ne sont pas ordonnés alors les résultats peuvent être aléatoires.

Pour un Stream parallèle, le fait que les éléments ne soient pas ordonnés peut améliorer les performances. Certaines opérations exécutées en parallèle sont plus efficaces si les éléments ne sont pas ordonnés : c'est par exemple le cas de l'opération distinct().

Il est possible d'utiliser l'opération unordered() pour retirer la caractéristique ordonnée des éléments d'un Stream. Dans certain cas, cela peut améliorer les performances de traitement de certaines opérations en parallèle sous réserve que l'ordre des éléments ne soit pas nécessaire.

Certaines opérations qui sont intrinsèquement liées à l'ordre, telles que limit(), peuvent nécessiter une mise en mémoire tampon pour respecter l'ordre, compromettant potentiellement l'intérêt de l'exécution des traitements en parallèle.

22.8.6. Les performances des traitements en parallèle

Il est très fréquent que l'utilisation d'un Stream parallèle soit décevante voire ne permette pas d'obtenir les performances attendues. Ceci est due à plusieurs raisons :

• L'amélioration des performances en parallélisant les traitements dépend du type de traitements exécutés et de la stratégie de parallélisation. La meilleure stratégie dépend du type de traitements
• La vitesse d'exécution est dépendante de l'environnement. Certains facteurs sont à prendre en compte : nombre de coeurs disponibles sur la machine, type de processeur, exécution dans un serveur d'applications, ...
• Pour obtenir une amélioration des performances, il est nécessaire d'avoir une quantité significative de données à traiter. La parallélisation des traitements nécessite un surcoût d'opérations (découpage en lots (fork) et d'agrégation des résultats (join)). Ce surcoût ne sera compensé que par le volume des données à traiter en parallèle
• Certains traitements peuvent voir leur performance s'effondrer en s'exécutant en parallèle. C'est notamment le cas, si ceux-ci induisent une forte contention en les parallélisant

Quel que soit le type de traitements à effectuer par le Stream en parallèle, la stratégie utilisée reposant sur le framework Fork/Join est toujours la même. Cette stratégie requiert :

• Un pool de threads, partagé par tout le code qui exécute des tâches avec Fork/Join
• Découper les données à traiter en sous-lots, éventuellement de manière récursive
• Exécuter les traitements sur un sous-lot de données par un des threads libres du pool
• Agréger les résultats intermédiaires de chaque sous-lots pour créer le résultat final

Evidemment, hormis l'exécution des traitements, toutes ces étapes ajoutent un surcoût au temps de traitement par rapport à une exécution séquentielle.

Ce mode de fonctionnement peut aussi induire des effets de bord liés au framework Fork/Join. Toutes les exécutions de code utilisant le framework Fork/Join se font par les threads du pool. Le nombre de ces threads est défini et donc limité.

Si une ou plusieurs tâches en cours d'exécution par le framework Fork/Join sont très longues, elles bloquent autant de threads du pool et ainsi peuvent induire une dégradation des performances liées à l'attente par certaines tâches de leur exécution par un thread libre. Cela concerne des traitements du Stream parallèle, d'autres Streams parallèles ou d'autres fonctionnalités du JDK ou personnelles qui utilisent aussi le framework Fork/Join (Parallel Arrays par exemple)

Par défaut, la taille du pool est égale au nombre de coeurs disponibles sur la machine moins un avec la valeur un comme minimum.

Si les traitements effectués par le Stream requièrent beaucoup de CPU, le gain de performance est contraint entre autres par le nombre de cours et par l'activité des autres threads sur la machine.

Si les traitements ont beaucoup de temps d'attente, le gain de performance peut aussi être plus perceptible.

L'utilisation d'un Stream en parallèle dans un serveur ou conteneur Java EE n'est généralement pas plus performant qu'un traitement en séquentiel. Pour permettre une meilleure montée en charge, les serveurs d'applications exécutent les traitements dans différents threads, aussi bien dans le conteneur web que dans le conteneur d'EJB. Rajouter une couche de parallélisation dans des traitements déjà en parallèle n'apporte aucune amélioration et engendre généralement une baisse des performances.

Pour s'assurer du gain de performance lors de l'utilisation d'un Stream en parallèle, il est fortement recommandé de faire des mesures sous la forme de benchmarks.

En cas de mesures des performances, il est nécessaire de les faire dans des conditions les plus poches possibles de la cible d'exploitation sur différents critères :

• Caractéristiques hardwares : type et nombre de CPU, type de disques durs, type et quantité de mémoire, ...
• Charge de traitements sur l'application mais aussi sur la machine elle-même le plus proche possible

Sans prendre ceci en compte, il est tout à fait probable que les performances soient bonnes en environnement de test ou de développement et mauvaises voire particulièrement mauvaises en production.

22.8.7. Des recommandations pour les Streams parallèles

Tous les traitements en parallèles exécutés par des Streams utilisent par défaut le pool de threads communs du framework Fork/Join. Il est donc nécessaire de ne pas exécuter en parallèle de longs traitements bloquants.

Lors de l'exécution des traitements d'un Stream en parallèle, l'ordre de traitement des éléments n'est pas garanti.

Il n'est pas recommandé d'utiliser du code qui produise des effets de bord dans les traitements d'un Stream. Ce type de traitement devrait généralement être fait différemment ou exécuté dans une boucle.

22.8.7.1. L'utilisation de traitements stateless

En programmation fonctionnelle, il est fortement recommandé de travailler de manière stateless : il ne faut pas modifier l'état des éléments en cours de traitement. Même si rien ne l'empêche, il ne faut pas modifier les éléments en cours de traitement par un Stream. C'est encore plus important dans le cas de l'exécution en parallèle des traitements sous peine d'avoir des comportements imprédictibles et des résultats erronés.

Le modèle de programmation des Streams est intrinsèquement stateless. L'utilisation d'expressions Lambda stateful dans des Streams exécutés en mode séquentiel ne pose généralement pas de soucis. Par contre, cette utilisation dans un Stream exécuté en mode parallèle est une très mauvaise idée qui conduit généralement à des problèmes pouvant engendrer des résultats aléatoires.

Tenter de modifier l'état mutable d'un objet partager dans les traitements d'un Stream implique des problématiques de concurrence d'accès. Les accès concurrents doivent être gérés manuellement pour éviter des problématiques de race conditions. La meilleure approche est de conserver tous les traitements du Stream stateless autant que possible. Cela implique généralement de revoir l'organisation des traitements.

Il n'est pas possible de compter sur l'API ou sur le compilateur pour s'assurer que les traitements soient stateless ou que si les traitements sont stateful alors les accès concurrents sont correctement gérés. Cette gestion implique obligatoirement une dégradation, plus ou moins importante, des performances liées généralement à l'ajout de contention.

Les Streams permettent l'exécution de leur pipeline d'opérations en parallèle à partir de diverses sources données qui ne sont pas forcement thread-safe comme par exemple une collection de type ArrayList. Pour que cela fonctionne correctement, il est impératif qu'il n'y ait pas de modifications des éléments de la source de données.

D'une manière générale, il est important de s'assurer que la source de données ne soit pas modifiée, de quelque manière que ce soit durant l'exécution du pipeline d'opérations d'un Stream. Cela est cependant possible, mais pas recommandé, si la source de données prend en charge la gestion des accès concurrents (le Spliterator associé au Stream doit avoir la caractéristique CONCURRENT).

22.8.7.2. Les effets de bord

Il est préférable de ne pas produire d'effets de bord dans les traitements fournis aux opérations d'un Stream afin de maintenir les traitements sans état et surtout d'éviter d'avoir des comportements inattendus lors de l'exécution des traitements en parallèle.

L'utilisation de traitements qui induisent des effets de bord dans des traitements en parallèle d'un Stream peuvent engendrer différentes difficultés, notamment :

• Il n'y a de garantie que les effets de bord soient vus par les autres threads
• L'utilisation d'objets mutables qui ne prennent pas en charge la gestion de la concurrence peut avoir des résultats non désirés
• L'utilisation d'objets synchronized induit de la contention qui peut engendrer des problèmes de performances

Lors de l'utilisation de Stream, il est tentant d'utiliser des opérations comme peek() ou forEach() pour manipuler un ou plusieurs objets mutables. La tentation est grande car ce type de traitement est utilisé dans les itérations externes réalisées avant Java 8. Bien sûre si ces opérations sont utilisées pour afficher des données, l'impact des effets de bords induits est négligeable. Par contre, si ces traitements modifient un objet mutable, il y a de forts risques d'avoir des résultats indésirés ou ayant des performances dégradées.

    ArrayList<Integer> resultats = new ArrayList<>();
    Stream.of(1, 2, 3, 4)
          .filter(i -> i % 2 == 0)
          .forEach(i -> resultats.add(i));
    System.out.println(resultats);

L'exemple ci-dessus produit un effet de bord en modifiant une collection de type List. Cette portion de code fonctionne correctement dans des traitements en séquentiel du Stream. Par contre, si les traitements sont exécutés en parallèle, et c'est très facile de modifier le code pour le faire, les résultats ne seront probablement pas ceux attendus car la classe ArrayList n'est pas thread-safe. Il est aussi possible dans ce cas d'utiliser une version synchronized de la collection mais cela va ajouter de la contention et donc dégrader les performances.

Il est préférable de ne pas utiliser de traitements qui effectuent des effets de bords. Dans l'exemple ci-dessus, c'est simple car il suffit d'utiliser un Collector pour aggéger les valeurs dans une collection.

    List<Integer> resultats = Stream.of(1, 2, 3, 4)
        .filter(i -> i % 2 == 0)
        .collect(Collectors.toList());
    System.out.println(resultats);

L'avantage de ce code est qu'il s'exécute correctement en séquentiel et surtout en parallèle aussi.

22.9. Les optimisations réalisées par l'API Stream

La manière de traiter les éléments par un Stream est particulière. On pourrait imaginer que les traitements sont exécutés de manière horizontale : tous les éléments sont traités dans leur intégralité par chaque opération les uns après les autres. Mais en fait, les traitements d'un Stream sont verticaux : chaque élément est traité successivement par chaque opération.

Ce comportement à l'exécution des traitements du Stream permet d'optimiser le nombre d'opérations à réaliser.

package fr.jmdoudoux.dej.streams;

import java.util.stream.Stream;

public class TestOperationsIntermediaires {

  public static void main(String[] args) {
    Stream.of("b2", "a1", "b1", "a2", "a3")
          .peek(e -> System.out.println("filter : " + e))
          .filter(e -> e.startsWith("a"))
          .peek(e -> System.out.println("anyMatch : " + e))
          .anyMatch(e -> e.contains("1"));
  }
}

filter: b2
filter: a1
anyMatch : a1

Dans l'exemple ci-dessus, l'opération filter n'est invoquée que sur deux éléments. Dès que le Predicat fourni à la méthode anyMatch() renvoie true pour un élément du Stream, les traitements de l'exécution du Stream sont interrompus. Cela permet d'optimiser le nombre d'opérations à exécuter pour obtenir le résultat.

22.9.1. Les optimisations liées aux opérations de type short circuiting

Les traitements réalisés en interne par l'API Steam sont conçus pour l'être de manière optimisée, en tout cas dans le cadre de traitements génériques.

Dans l'exemple ci-dessous, trois opérations intermédiaires sont appliquées. Celles-ci ne sont pas appliquées sur tous les éléments de la source.

package fr.jmdoudoux.dej.streams;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class Test {

  public static void main(String[] args) {
    List<Personne> personnes = new ArrayList<>(6);
    personnes.add(new Personne("p1", Genre.HOMME, 176));
    personnes.add(new Personne("p2", Genre.HOMME, 190));
    personnes.add(new Personne("p3", Genre.FEMME, 182));
    personnes.add(new Personne("p4", Genre.FEMME, 162));
    personnes.add(new Personne("p5", Genre.HOMME, 186));
    personnes.add(new Personne("p6", Genre.FEMME, 168));

    List<String> nomsDeuxPlusGrands = personnes.stream().filter(p -> {
      System.out.println("filter - " + p);
      return p.getTaille() > 180;
    }).map(p -> {
      System.out.println("map - " + p);
      return p.getNom();
    }).limit(2).collect(Collectors.toList());

    System.out.println("Resultat : " + nomsDeuxPlusGrands);
  }
}

filter - Personne [nom=p1, genre=HOMME, taille=176]
filter - Personne [nom=p2, genre=HOMME, taille=190]
map - Personne [nom=p2, genre=HOMME, taille=190]
filter - Personne [nom=p3, genre=FEMME, taille=182]
map - Personne [nom=p3, genre=FEMME, taille=182]
Resultat : [p2, p3]

Le filtre n'est appliqué que sur 3 éléments et la transformation uniquement sur 2 éléments

Les autres éléments ne sont plus traités une fois que l'opération limit() de type short-circuiting est satisfaite.

Les opérations intermédiaires possèdent une caractéristique intéressante : elles sont lazy.

package fr.jmdoudoux.dej.streams;

import java.util.stream.Stream;

public class TestOperationsIntermediaires {

  public static void main(String[] args) {
    Stream.of("b2", "a1", "b1", "a2", "a3")
          .filter(e -> {System.out.println("filter : " + e);
                        return e.startsWith("a");
                       });
  }
}

Lors de l'exécution de cet exemple, aucun message n'est affiché à la console.

L'exécution des traitements des opérations n'est déclenchée que lors de l'invocation de la méthode terminale.

package fr.jmdoudoux.dej.streams;

import java.util.stream.Stream;

public class TestOperationsIntermediaires {

  public static void main(String[] args) {
    Stream.of("b2", "a1", "b1", "a2", "a3")
          .peek(e -> System.out.println("filter : " + e))
          .filter(e -> e.startsWith("a"))
          .forEach(e -> System.out.println("foreach : " + e));
  }
}

filter : b2
filter : a1
foreach : a1
filter : b1
filter : a2
foreach : a2
filter : a3
foreach : a3

22.9.2. L'ordre des opérations d'un Stream

L'ordre des opérations définies dans le pipeline peut avoir un impact non négligeable sur les performances d'exécution des traitements.

package fr.jmdoudoux.dej.streams;

import java.util.stream.Stream;

public class TestOperationsIntermediaires {

  public static void main(String[] args) {
    Stream.of("b2", "a1", "b1", "a2", "a3")
          .peek(e -> System.out.println("map : " + e))
          .map(String::toUpperCase)
          .peek(e -> System.out.println("filter : " + e))
          .filter(e -> e.startsWith("B"))
          .forEach(System.out::println);
  }
}

map: b2
filter: B2
B2
map: a1
filter: A1
map: b1
filter: B1
B1
map: a2
filter: A2
map: a3
filter: A3

Dans l'exemple ci-dessus, les opérations sont invoquées sur tous les éléments du Stream.

Il est possible d'optimiser ces traitements simplement en intervertissant les opérations pour d'abord filter les éléments et ainsi réduire le nombre d'éléments sur laquelle la transformation est exécutée.

package fr.jmdoudoux.dej.streams;

import java.util.stream.Stream;

public class TestOperationsIntermediaires {

  public static void main(String[] args) {
    Stream.of("b2", "a1", "b1", "a2", "a3")
          .peek(e -> System.out.println("filter : " + e))
          .filter(e -> e.startsWith("b"))
          .peek(e -> System.out.println("map : " + e))
          .map(String::toUpperCase)
          .forEach(System.out::println);
  }
}

filter : b2
map : b2
B2
filter : a1
filter : b1
map : b1
B1
filter : a2
filter : a3

C'est encore plus important lorsqu'une opération stateful est utilisée. L'exemple ci-dessous trie les éléments puis les filtre et les transforme pour les afficher.

package fr.jmdoudoux.dej.streams;

import java.util.stream.Stream;

public class TestOperationsIntermediaires {

  public static void main(String[] args) {
    Stream.of("b2", "a1", "b1", "a2", "a3")
          .sorted((e1, e2) -> {
            System.out.printf("sort: %s; %s\n", e1, e2);
            return e1.compareTo(e2);
          })
          .peek(e -> System.out.println("filter : " + e))
          .filter(e -> e.startsWith("b"))
          .peek(e -> System.out.println("map : " + e))
          .map(String::toUpperCase)
          .forEach(System.out::println);
  }
}

sort: a1; b2
sort: b1; a1
sort: b1; b2
sort: b1; a1
sort: a2; b1
sort: a2; a1
sort: a3; b1
sort: a3; a2
filter: a1
filter: a2
filter: a3
filter: b1
map: b1
B1
filter: b2
map: b2
B2

L'opération de tri requiert la comparaison successive d'éléments deux à deux pour assurer ses traitements. Dans ce cas, il est possible d'optimiser les traitements en effectuant d'abord l'opération de filtre et ainsi réduire le nombre d'éléments à trier.

package fr.jmdoudoux.dej.streams;

import java.util.stream.Stream;

public class TestOperationsIntermediaires {

  public static void main(String[] args) {
    Stream.of("b2", "a1", "b1", "a2", "a3")
          .peek(e -> System.out.println("filter : " + e))
          .filter(e -> e.startsWith("b"))
          .sorted((e1, e2) -> {
            System.out.printf("sort: %s; %s\n", e1, e2);
            return e1.compareTo(e2);
          })
          .peek(e -> System.out.println("map : " + e))
          .map(String::toUpperCase)
          .forEach(System.out::println);
  }
}

filter: b2
filter: a1
filter: b1
filter: a2
filter: a3
sort: b1; b2
map: b1
B1
map: b2
B2

22.10. Les Streams infinis

Un Stream infini fait référence à un Stream dont la source produit de manière continue des éléments à consommer. Sans condition d'arrêt, cette génération peut être infinie comme son l'indique.

La mise en oeuvre de Stream infini est possible car le principe de traitements des éléments d'un Stream est lazy. Les opérations intermédiaires définissent les traitements mais ceux-ci ne sont réellement exécutés que lors de l'invocation de l'opération terminale.

Stream<Integer> stream = Stream.iterate(0, i -> i + 1);

Attention lors de l'utilisation de Stream infinis : il est nécessaire de fournir une restriction qui va limiter le nombre d'éléments générés sinon le Stream va exécuter sans arrêt la génération de nouveaux éléments.

// attention : ce code effectue une boucle infinie

IntStream.iterate(0, i -> i + 1).forEach(System.out::println);

L'exemple ci-dessous induit une boucle infinie.

Il faut impérativement introduire une condition d'arrêt par exemple en utilisant l'opération limit() qui permet de limiter le nombre d'éléments contenus dans le Stream.

IntStream.iterate(0, i -> i + 1).limit(5).forEach(System.out::println);

Parfois les traitements infinis sont plus subtils et bien qu'il existe une condition d'arrêt, il est nécessaire de s'assurer que celle-ci sera vérifiée tôt ou tard.

List<Double> valeur = Stream
    .generate(Math::random)
    .filter(v -> (v > 10) && (v < 20))
    .limit(10)
    .collect(Collectors.toList());

Dans l'exemple ci-dessus, les traitements du Stream ne s'arrêtent jamais bien qu'une opération limit(10) soit utilisée. Le problème vient du fait que la méthode random() de la classe Math renvoie une valeur comprise entre 0 et 1. Hors le filtre ne garde que les valeurs comprises entre 10 et 20 exclues. Aucune valeur générée n'est donc conservée par le filtre et la limite de 10 n'est jamais atteinte : ceci provoque une boucle infinie qui génère des nombres aléatoires.

Dans un Stream parallèle, c'est la catastrophe car tous les processeurs sont sollicités pour cette boucle infinie.

List<Double> valeur = Stream
    .generate(Math::random)
    .parallel()
    .filter(v -> (v > 10) && (v < 20))
    .limit(10)
    .collect(Collectors.toList());

Même si ce n'est pas son but, il est possible d'utiliser un Stream infini pour réaliser l'équivalent d'une boucle. L'opération limit() permet de préciser le nombre d'itération qui sera réalisée.

Stream.iterate(0, i -> i + 1)
      .limit(10)
      .forEach(System.out::println);

Cet exemple est équivalent à l'exemple ci-dessous

i = 0;
while (i < 10) {
  System.out.println(i);
  i++;
}

Attention : dans des cas aussi basique que celui présenté ci-dessus, l'utilisation d'un Stream est moins performante et consomme plus de ressources.

L'opération generate() permet aussi de créer un Stream infini en lui passant en paramètre un Supplier qui aura la responsabilité de créer de nouveaux éléments.

Stream<UUID> streamUUID = Stream.generate(UUID::randomUUID);

Comme avec la méthode iterate(), il est nécessaire de fournir une condition d'arrêt à la génération de nouveaux éléments par le Stream.

List<UUID> valeurs = streamUUID
  .limit(5)
  .collect(Collectors.toList());

22.11. Le débogage d'un Stream

Le débogage d'un Stream n'est pas simple car une majorité des traitements est réalisée en interne par l'API.

Lorsqu'une anomalie survient dans les traitements d'un Stream, la stacktrace n'est généralement pas d'une grande aide.

import java.util.stream.Stream;

public class DebugStream {

  public static void main(String[] args) {
    Double tailleMoyenne = Stream.of("texte", null, "grand texte")
       .mapToInt(String::length)
        .average()
        .getAsDouble();
    System.out.println("taille moyenne = " + tailleMoyenne);
  }
}

Exception in thread "main" java.lang.NullPointerException
      at java.util.stream.ReferencePipeline$4$1.accept(ReferencePipeline.java:210)
      at java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:175)
      at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
      at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
      at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
      at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708)
      at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
      at java.util.stream.IntPipeline.collect(IntPipeline.java:472)
      at java.util.stream.IntPipeline.average(IntPipeline.java:434)
      at DebugStream.main(DebugStream.java:14)

Une exception de type NullPointerException est levée mais la lecture de la stacktrace ne fournit aucune information pour permettre de déterminer l'origine du problème.

Il est possible d'utiliser la méthode peek() pour afficher l'élément en cours de traitement dans le pipeline. C'est ailleurs normalement la seule raison d'utiliser la méthode peek() dans un pipeline d'opérations. Pour simplement afficher l'élément courant, il suffit de lui passer en paramètre la référence de méthode System.out::println. Si l'on doit utiliser la méthode peek() plusieurs fois dans le pipeline, il est préférable de construire une chaîne de caractères qui fournisse des informations complémentaires notamment l'étape courante dans le pipeline.

import java.util.stream.Stream;

public class DebugStream {

  public static void main(String[] args) {
    Double tailleMoyenne = Stream.of("texte", null, "grand texte")
        .peek(e ->System.out.println("map : " + e))
        .mapToInt(String::length)
        .peek(e -> System.out.println("average : " + e))
        .average()
        .getAsDouble();
    System.out.println("taille moyenne = " + tailleMoyenne);
  }
}

map : texte
average : 5
map : null
Exception in thread "main" java.lang.NullPointerException
      at java.util.stream.ReferencePipeline$4$1.accept(ReferencePipeline.java:210)
      at java.util.stream.ReferencePipeline$11$1.accept(ReferencePipeline.java:373)
      at java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:175)
      at java.util.stream.ReferencePipeline$11$1.accept(ReferencePipeline.java:373)
      at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
      at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
      at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
      at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708)
      at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
      at java.util.stream.IntPipeline.collect(IntPipeline.java:472)
      at java.util.stream.IntPipeline.average(IntPipeline.java:434)
      at DebugStream.main(DebugStream.java:14)

Si le problème est dans le code d'une expression lambda fournie en paramètre d'une opération, il est possible d'effectuer un refactoring pour inclure le code de l'expression dans une méthode et passer en paramètre de l'opération une référence de méthode permettant son invocation. Il suffit alors de mettre un point d'arrêt dans la méthode.

Les principaux IDE permettent de mettre des points d'arrêts pour faciliter le débogage du code fourni dans les expressions Lambda des opérations. Mais parfois cela ne suffit pas pour permettre de déboguer les traitements d'un Stream.

22.12. Les limitations de l'API Stream

L'API Stream permet la mise en oeuvre d'une approche fonctionnelle dans le langage Java. Java reste un langage orienté objet et n'est pas un langage fonctionnel. L'utilisation de l'API Stream n'est pas aussi riche ou poussée que dans d'autres langages qui sont fonctionnels.

L'API Stream possède aussi quelques limitations. Par exemple, il n'est pas possible de définir ses propres opérations : seules celles définies par l'API peuvent être utilisées.

22.12.1. Un Stream n'est pas réutilisable

Une fois qu'un Stream a été exécuté, il ne peut plus être réutilisé. Dès qu'une opération terminale est invoquée, celle-ci ferme le Stream.

Si le Stream est de nouveau invoqué, une exception de type IllegalStateException est levée.

package fr.jmdoudoux.dej.streams;

import java.util.stream.Stream;

public class TestStream {

  public static void main(String[] args) {
    Stream<String> stream = Stream.of("a1", "a2", "a3")
                                  .filter(s -> s.startsWith("a"));
    stream.forEach(System.out::println);
    stream.forEach(System.out::println);
  }
}

a1
a2
a3
Exception in thread "main" java.lang.IllegalStateException: stream has already been
operated upon or closed
      at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:229)
      at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418)
      at fr.jmdoudoux.dej.streams.TestStream.main(TestStream.java:11)

Pour contourner cette limitation, il faut créer un nouveau Stream pour chaque utilisation. Si la configuration du Stream est complexe, il est possible de définir un Supplier dont le rôle sera de créer une instance à chaque invocation.

package fr.jmdoudoux.dej.streams;

import java.io.PrintStream;
import java.util.function.Supplier;
import java.util.stream.Stream;

public class TestStream {

  public static void main(String[] args) {
    Supplier<Stream<String>> supplier = () -> Stream.of("a1", "a2", "a3")
                    .filter(s -> s.startsWith("a"));
    supplier.get().forEach(System.out::println);
    supplier.get().forEach(System.out::println);
  }
}

a1
a2
a3
a1
a2
a3

22.13. Quelques recommandations sur l'utilisation de l'API Stream

L'API Stream permet de réaliser certains traitements sur un ensemble de données de manière déclarative, ce qui réduit la quantité de code à produire. La déclaration de certains traitements peut se faire de différentes manières pouvant avoir des impacts sur les performances.

Il ne faut pas utiliser systématiquement l'API Stream mais plutôt favoriser son utilisation lorsque celle-ci apporte une plus-value. Si l'API est utilisée, il est préférable de le faire de manière optimale.

22.13.1. L'utilisation à mauvais escient de l'API Stream

Parfois certaines utilisations de l'API Stream peuvent rendre le code moins lisible et compréhensible ou moins performant.

22.13.1.1. Le parcours des éléments

Il n'est pas utile d'avoir recours à un Stream pour appliquer un traitement sur les éléments d'une collection. C'est d'autant plus vrai si aucune opération intermédiaire n'est utilisée dans le Stream.

import java.util.Arrays;
import java.util.List;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine", "pierre" };
    List<String> prenoms = Arrays.asList(prenomsTab);
    prenoms.stream()
           .forEach(System.out::println);
  }
}

Il est préférable d'utiliser la méthode forEach() sur la collection sans utiliser un Stream.

import java.util.Arrays;
import java.util.List;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine", "pierre" };
    List<String> prenoms = Arrays.asList(prenomsTab);
    prenoms.forEach(System.out::println);
  }
}

L'utilisation de cette méthode garantie l'ordre des éléments lors du parcours si la collection est ordonnée.

22.13.1.2. Le remplacement des boucles for par un Stream

Il est courant de voir le remplacement des boucles for par un Stream de manière un peu systématique.

import java.util.stream.IntStream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    for (int i = 0; i < 5; i++) {
      System.out.println(i);
    }
    IntStream.range(0, 5)
             .forEach(System.out::println);
  }
}

Evidemment cet exemple est très (trop) simple. L'exemple ci-dessous qui imbrique deux itérations est probablement plus lisible et donc maintenable avec des boules for qu'avec leur équivalent utilisant des Streams

import java.util.stream.IntStream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    System.out.println("Tables de multiplications");
    for (int i = 0; i < 10; i++) {
      for (int j = 0; j < 10; j++)
        System.out.format("%2d ", i * j);
      System.out.println();
    }
    System.out.println("Tables de multiplications");
    IntStream.range(0, 10)
             .forEach(i -> {
                 IntStream.range(0, 10)
                          .forEach(j -> { System.out.format("%2d ", i * j); });
               System.out.println();
             });
  }
}

Si les traitements de la boucle for sont simples ou stateful ou qu'il n'est pas prévu de les paralléliser alors généralement leur remplacement par un Stream rend le code parfois moins maintenable et/ou moins performant.

Le remplacement d'une boucle for par un Stream n'est réellement intéressant que si l'approche fonctionnelle s'appuyant sur une itération interne est utilisée pour par exemple chaîner plusieurs opérations dans le but d'obtenir un résultat.

Ceci ne concerne pas forcément que l'API Stream : c'est généralement le cas lorsqu'on remplace du code de base par l'utilisation d'un framework. On gagne en fonctionnalité et en souplesse mais on perd en performance, d'autant plus si aucune attention particulière n'est prêtée lors de la mise en oeuvre du framework.

Une autre raison est la clarté des stacktraces si une exception est levée durant les traitements des itérations. Celles-ci sont très claires dans le cas d'une boucle for.

public class StreamBonnePratique {

  public static void main(String[] args) {
    for (int i = 0; i < 10; i++) {
      System.out.println(i / (9 - i));
    }
  }
}

Exception in thread "main" java.lang.ArithmeticException: / by zero
      atStreamBonnePratique.main(StreamBonnePratique.java:6)

Elles sont plus verbeuses dans le cas d'un Stream

import java.util.stream.IntStream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    IntStream.range(0, 10)
             .forEach(i -> {
                System.out.println(i / (9 - i));
             });
  }
}

Exception in thread "main" java.lang.ArithmeticException: / by zero
      at StreamBonnePratique.lambda$0(StreamBonnePratique.java:9)
      at java.util.stream.Streams$RangeIntSpliterator.forEachRemaining(Streams.java:110)
      at java.util.stream.IntPipeline$Head.forEach(IntPipeline.java:557)
      at StreamBonnePratique.main(StreamBonnePratique.java:8)

22.13.1.3. La conversion d'une collection

Il n'est pas utile d'avoir recours à un Stream pour convertir les éléments d'un tableau en une collection de type List.

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine", "pierre" };
    List<String> prenoms = Stream.of(prenomsTab)
                                 .collect(Collectors.toList());
  }
}

Il est préférable d'utiliser la méthode toList() de la classe Arrays.

import java.util.Arrays;
import java.util.List;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine", "pierre" };
    List<String> prenoms = Arrays.asList(prenomsTab);
  }
}

De la même manière, il n'est pas nécessaire d'utiliser un Stream pour convertir une collection vers une autre collection.

import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine", "pierre" };
    List<String> prenomsList = Arrays.asList(prenomsTab);
    Set<String> prenoms = prenomsList.stream()
                                     .collect(Collectors.toSet());
  }
}

Il est préférable d'utiliser la surcharge du constructeur des collections qui attend en paramètre une collection.

import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine",  "pierre" };
    List<String> prenomsList = Arrays.asList(prenomsTab);
    Set<String> prenoms = new HashSet<>(prenomsList);
  }
}

Enfin, il n'est pas nécessaire d'utiliser un Stream pour convertir une collection en un tableau.

import java.util.Arrays;
import java.util.List;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine", "pierre" };
    List<String> prenomsList = Arrays.asList(prenomsTab);
    String[] prenoms = prenomsList.stream()
                                  .toArray(String[]::new);
  }
}

Il est préférable d'utiliser la méthode toArray() de l'interface Collection.

import java.util.Arrays;
import java.util.List;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine", "pierre" };
    List<String> prenomsList = Arrays.asList(prenomsTab);
    String[] prenoms = prenomsList.toArray(new String[0]);
  }
}

22.13.1.4. La recherche du plus grand élément d'une collection

Il n'est pas nécessaire d'utiliser un Stream pour uniquement trouver le plus grand élément dans une collection.

import java.util.Arrays;
import java.util.Comparator;
import java.util.List;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine", "pierre" };
    List<String> prenomsList = Arrays.asList(prenomsTab);
    String plusGrand = prenomsList.stream()
                                  .max(Comparator.naturalOrder())
                                  .orElse(null);
    System.out.println(plusGrand);
  }
}

Il est préférable d'utiliser la méthode max() de la classe Collections.

import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine", "pierre" };
    List<String> prenomsList = Arrays.asList(prenomsTab);
    String plusGrand = prenomsList.isEmpty() ? null 
       : Collections.max(prenomsList, Comparator.naturalOrder());
    System.out.println(plusGrand);
  }
}

22.13.1.5. La détermination du nombre d'éléments d'une collection

Il n'est pas nécessaire d'utiliser un Stream pour uniquement déterminer le nombre d'éléments d'une collection.

import java.util.Arrays;
import java.util.List;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine", "pierre" };
    List<String> prenomsList = Arrays.asList(prenomsTab);
    long nbElements = prenomsList.stream()
                                 .count();
  }
}

Il est préférable d'utiliser la méthode size() de l'interface Collection.

import java.util.Arrays;
import java.util.List;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine", "pierre" };
    List<String> prenomsList = Arrays.asList(prenomsTab);
    long nbElements = prenomsList.size();
  }
}

22.13.1.6. La vérification de la présence d'un élément dans une collection

Il n'est pas nécessaire d'utiliser un Stream pour uniquement vérifier la présence d'un élément dans une collection.

import java.util.Arrays;
import java.util.List;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine", "pierre" };
    List<String> prenomsList = Arrays.asList(prenomsTab);
    boolean trouve = prenomsList.stream()
                                .anyMatch(s -> "pierre".equals(s));
  }
}

Il est préférable d'utiliser la méthode contains() de l'interface Collection.

import java.util.Arrays;
import java.util.List;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine", "pierre" };
    List<String> prenomsList = Arrays.asList(prenomsTab);
    boolean trouve = prenomsList.contains("pierre");
  }
}

Ceci est particulièrement vrai pour une collection de type Set.

22.13.2. L'utilisation de l'API Stream pour rechercher des éléments

Certaines recherches d'éléments peuvent parfois se faire de différentes manières. Certaines sont plus intéressantes d'un point de vue performance et/ou maintenabilité.

22.13.2.1. La recherche de la présence d'un élément

Il n'est pas utile de filtrer les éléments et de vérifier s'il y a un premier élément restant pour s'assurer qu'un élément est présent dans une collection.

mport java.util.Arrays;
import java.util.List;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine", "pierre" };
    List<String> prenomsList = Arrays.asList(prenomsTab);
    boolean trouve = prenomsList.stream()
                                .filter(s -> "pierre".equals(s))
                                .findFirst()
                                .isPresent();
  }
}

Il est préférable d'utiliser l'opération terminale anyMatch() du Stream.

mport java.util.Arrays;
import java.util.List;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine", "pierre" };
    List<String> prenomsList = Arrays.asList(prenomsTab);
    boolean trouve = prenomsList.stream()
                                .anyMatch(s-> "pierre".equals(s));
  }
}

22.13.2.2. La recherche du plus petit élément

Il n'est pas utile de trier les éléments dans un ordre croissant et de prendre le premier pour rechercher le plus petit élément.

import java.util.Comparator;
import java.util.Optional;
import java.util.stream.Stream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    Stream<String> prenoms = Stream.of("pierre", "anne", "sophie", "thierry", "antoine");
    Optional<String> premier = prenoms.sorted()
                                      .findFirst();
    System.out.println(premier.orElse("inconnu"));
  }
}

Il est préférable d'utiliser l'opération min() du Stream.

import java.util.Optional;
import java.util.stream.Stream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    Stream<String> prenoms = Stream.of("pierre", "anne", "sophie", "thierry", "antoine");
    Optional<String> premier = prenoms.min(Comparator.naturalOrder());
    System.out.println(premier.orElse("inconnu"));
  }
}

22.13.2.3. La recherche du plus grand élément

Il n'est pas utile de trier les éléments dans un ordre décroissant et de prendre le premier pour rechercher le plus grand élément.

import java.util.Comparator;
import java.util.Optional;
import java.util.stream.Stream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    Stream<String> prenoms = Stream.of("pierre", "anne", "sophie", "thierry", "antoine");
    Optional<String> premier = prenoms.sorted(Comparator.reverseOrder())
                                     .findFirst();
    System.out.println(premier.orElse("inconnu"));
  }
}

Il est préférable d'utiliser l'opération max() du Stream.

import java.util.Comparator;
import java.util.Optional;
import java.util.stream.Stream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    Stream<String> prenoms = Stream.of("pierre", "anne", "sophie", "thierry", "antoine");
    Optional<String> premier = prenoms.max(Comparator.naturalOrder());
    System.out.println(premier.orElse("inconnu"));
  }
}

22.13.3. La création de Streams

Il est préférable d'utiliser les fabriques dédiées de l'API Stream plutôt que de créer une collection à partir de laquelle on obtient un Stream pour traiter les éléments.

22.13.3.1. La création d'un Stream vide

Il n'est pas utile de créer une collection vide et de demander un Stream sur celle-ci pour obtenir un Stream vide.

import java.util.Collections;
import java.util.stream.Stream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    Stream<Object> streamVide = Collections.emptyList()
                                           .stream();
  }
}

Il est préférable d'utiliser la méthode empty() de l'interface Stream.

import java.util.stream.Stream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    Stream<Object> streamVide = Stream.empty();
  }
}

22.13.3.2. La création d'un Stream avec un seul élément

Il n'est pas utile de créer une collection avec un seul élément de type Set (avec la méthode singleton()) ou List (avec la méthode singletonList()) et de demander un Stream sur celle-ci pour obtenir un Stream.

import java.util.Collections;
import java.util.stream.Stream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    Stream<String> stream = Collections.singleton("test")
                                       .stream();
  }
}

Il est préférable d'utiliser la méthode of() de l'interface Stream.

import java.util.stream.Stream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    Stream<String> stream = Stream.of("test");
  }
}

22.13.3.3. La création d'un Stream à partir d'un tableau

Il n'est pas utile de créer une collection avec les éléments du tableau et demander un Stream sur celle-ci pour obtenir un Stream.

import java.util.Collections;
import java.util.stream.Stream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine", "pierre" };
    Stream<String> prenoms = Arrays.asList(prenomsTab).stream();
  }
}

Il est préférable d'utiliser la méthode of() de l'interface Stream ou la méthode stream() de la classe Arrays.

import java.util.stream.Stream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine", "pierre" };
    Stream<String> prenoms = Stream.of(prenomsTab);
    Stream<String> prenoms2 =Arrays.stream(prenomsTab);
  }
}

22.13.3.4. Le traitement d'une plage d'éléments d'un tableau

Il est pas utile d'utiliser la fabrique range() de l'interface IntStream puis de transformer chaque valeur en l'élément correspondant dans le tableau avec l'opération mapToObj().

import java.util.stream.IntStream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine", "pierre" };
    long nb = IntStream.range(2, 5)
                       .mapToObj(idx -> prenomsTab[idx])
                       .count();
    System.out.println(nb);
  }
}

Il est préférable d'utiliser la surcharge de la méthode stream() de la classe Arrays qui attend en paramètre le tableau, l'index de début et l'index de fin de la plage des éléments à traiter.

import java.util.Arrays;

public class StreamBonnePratique {

  public static void main(String[] args) {
    String[] prenomsTab = { "alain", "anne", "sophie", "thierry", "antoine", "pierre" };
    long nb = Arrays.stream(prenomsTab, 2, 5)
                    .count();
    System.out.println(nb);
  }
}

22.13.4. L'utilisation non requise d'un Collector

Certains Collector sont conçus pour être utilisés comme des downstreams Collector de l'opération groupingBy(). Il n'est pas nécessaire de les utiliser à la place des opérations de l'interface Stream équivalente.

Par exemple, pour compter simplement les éléments d'un Stream, il n'est pas utile d'utiliser le Collector obtenu par la fabrique counting().

import java.util.stream.Collectors;
import java.util.stream.Stream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    Stream<String> prenoms = Stream.of("pierre", "anne", "sophie", "thierry", "antoine");
    long nb = prenoms.collect(Collectors.counting());
    System.out.println(nb);
  }
}

Il est préférable d'utiliser l'opération count() de l'interface Stream.

import java.util.stream.Stream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    Stream<String> prenoms = Stream.of("pierre", "anne", "sophie", "thierry", "antoine");
    long nb = prenoms.count();
    System.out.println(nb);
  }
}

De même, pour déterminer le plus petit ou le plus grand élément d'un Stream, il n'est pas utile d'utiliser le Collector obtenu par la fabrique minBy() ou maxBy().

import java.util.Comparator;
import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    Stream<String> prenoms = Stream.of("pierre", "anne", "sophie", "thierry", "antoine");
    Optional<String> plusGrand = prenoms.collect(Collectors.maxBy(Comparator.naturalOrder()));
    System.out.println(plusGrand.orElse("inconnu"));
  }
}

Il est préférable d'utiliser l'opération min() ou max() de l'interface Stream.

import java.util.Comparator;
import java.util.Optional;
import java.util.stream.Stream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    Stream<String> prenoms = Stream.of("pierre", "anne", "sophie", "thierry", "antoine");
    Optional<String> plusGrand = prenoms.max(Comparator.naturalOrder());
    System.out.println(plusGrand.orElse("inconnu"));
  }
}

De la même manière, il est préférable d'utiliser :

• L'opération reduce() plutôt que le Collector obtenu par la méthode reducing() de la classe Collectors
• L'opération map() plutôt que le Collector obtenu par la méthode mapping() de la classe Collectors
• L'opération filter() plutôt que le Collector obtenu par la méthode filtering() de la classe Collectors
• L'opération flatMap() plutôt que le Collector obtenu par la méthode flatMapping() de la classe Collectors

22.13.5. Le traitement de valeurs numériques

Pour éviter des opérations d'autoboxing et donc d'améliorer les performances, il n'est pas recommandé d'effectuer des calculs sur des wrappers de type primitif numérique.

import java.util.stream.Stream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    Personne p1 = new Personne("alain", 179);
    Personne p2 = new Personne("sopihe", 176);
    Personne p3 = new Personne("thierry", 172);

    int somme = Stream.of(p1, p2, p3)
                      .map(p -> p.getTaille())
                      .reduce(0, (a, b) -> a + b);
    System.out.println(somme);
  }
}

Il est préférable d'utiliser un IntStream, LongStream ou DoubleStream pour améliorer les performances, réduire le nombre d'objets créés et réduire la quantité de code nécessaire.

import java.util.stream.Stream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    Personne p1 = new Personne("alain", 179);
    Personne p2 = new Personne("sopihe", 176);
    Personne p3 = new Personne("thierry", 172);

    int somme = Stream.of(p1, p2, p3)
                      .mapToInt(p -> p.getTaille())
                      .sum();
    System.out.println(somme);
  }
}

22.13.6. Ne compter les éléments que si c'est nécessaire

Certains traitements exécutés en comptant le nombre d'éléments sont assez peu performants et il est parfois préférable d'utiliser une autre approche notamment si le nombre d'éléments est important.

22.13.6.1. La détermination du nombre d'éléments de sous-ensembles

Par exemple : des groupes contiennent des personnes et on souhaite calculer le nombre de personnes incluses dans un ensemble de groupes.

Une première approche consiste à mettre à plat chaque personne de chaque groupe en utilisant l'opération flatMap() et compter le nombre de personnes obtenues.

import java.util.Arrays;
import java.util.stream.Stream;

public class StreamBonnePratique {

  public static void main(String[] args) {
    Personne p1 = new Personne("Alain", 179);
    Personne p2 = new Personne("Sophie", 176);
    Personne p3 = new Personne("Thierry", 172);
    Personne p4 = new Personne("Martine", 169);
    Personne p5 = new Personne("Paul", 174);
    Personne p6 = new Personne("Isabelle", 182);
    Personne p7 = new Personne("Jean", 172);

    Groupe g1 = new Groupe("Groupe1", Arrays.asList(p1, p2, p3));
    Groupe g2 = new Groupe("Groupe2", Arrays.asList(p4, p5));
    Groupe g3 = new Groupe("Groupe3", Arrays.asList(p6, p7));

    long somme = Stream.of(g1, g2, g3)
                       .flatMap(g -> g.getPersonnes()
                                      .stream())
                       .count();
    System.out.println(somme);
  }
}