5. Il tipo Stream<T> di Java 8
5.1. Esempio-01 - la classe Stream
Le operazioni sui flussi Observable presentano numerosi punti in comune con i flussi Stream. Una differenza è che un elemento di uno stream Stream non può essere elaborato prima che l'intero stream Stream sia stato ottenuto, mentre un elemento di uno stream Observable può essere elaborato (osservato) non appena viene ricevuto, senza attendere il completamento dell’intero flusso Observable. Un’altra differenza è che, una volta ottenuto il flusso Stream, i suoi valori vengono estratti (pull) uno alla volta dal flusso Stream. Per l’osservabile, invece, il funzionamento è diverso: non appena emette un valore, questo viene inviato (pushed) al suo abbonato.
Diverse classi implementano il concetto di Stream. Qui presentiamo la classe Stream<T>:
La classe Stream dispone di 39 metodi. Ne presenteremo alcuni. Consideriamo il codice seguente:
 |
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple01 {
public static void main(String[] args) {
// elenco delle persone
List<Personne> personnes = Personnes.get();
// visualizzazione 1
personnes.stream().forEach(p -> {
System.out.println(p);
});
System.out.println("----------------");
// visualizzazione 2
personnes.stream().forEach(System.out::println);
}
}
- riga 11: si istanzia un elenco di persone;
- riga 13: a partire da questo elenco, si crea un Stream. Tutte le collezioni possono essere così trasformate in flussi Stream. Ciò consente di sfruttare tutti i metodi di questa classe, che permettono di elaborare gli elementi della collezione in modo più conciso rispetto ai cicli. Consente inoltre di beneficiare del parallelismo nell’elaborazione degli elementi, quando possibile;
- riga 13: il metodo [Stream.forEach] ha la seguente firma:
 |
Si nota che il parametro del metodo è l’interfaccia funzionale [Consumer<T>] presentata al paragrafo 4.4, un’interfaccia il cui unico metodo utilizza il tipo T e non restituisce alcun valore.
- nel codice:
personnes.stream().forEach(p -> {
System.out.println(p);
});
- [personnes.stream()] produce un flusso di elementi di tipo [Personne] che alimenta il metodo [forEach]. Il parametro p è di tipo [Personne] e la funzione lambda fornita visualizza questa persona;
Il codice precedente può essere semplificato come segue (riga 18):
personnes.stream().forEach(System.out::println);
Anziché passare come parametro il valore di una funzione lambda, passiamo il riferimento a un metodo esistente, in questo caso il metodo println della classe System.out. Ovviamente questo metodo deve avere la firma corretta, in questo caso la firma del metodo [Consumer.accept]: void accept(T t). Come detto in precedenza, il parametro del metodo [accept] sarà di tipo [Personne];
Otteniamo i seguenti risultati:
Una volta che uno Stream è stato utilizzato, non è più utilizzabile. È necessario ricostruirlo se si desidera utilizzarlo nuovamente. Ciò è illustrato dal seguente codice [Exemple01b]:
package dvp.java8.streams;
import java.util.stream.Stream;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple01b {
public static void main(String[] args) {
// flusso di persone
Stream<Personne> personnes = Personnes.get().stream();
// visualizzazione 1
personnes.forEach(p -> {
System.out.println(p);
});
System.out.println("----------------");
// visualizzazione 2
personnes.forEach(System.out::println);
}
}
- riga 11: per ottimizzare il codice, si decide di costruire il Stream una sola volta. I risultati ottenuti sono quindi i seguenti:
{"nom":"jean","age":20,"poids":70.0,"sexe":"HOMME"}
{"nom":"marie","age":10,"poids":30.0,"sexe":"FEMME"}
{"nom":"camille","age":30,"poids":55.0,"sexe":"FEMME"}
----------------
Exception in thread "main" java.lang.IllegalStateException: stream has already been operated upon or closed
at java.util.stream.AbstractPipeline.sourceStageSpliterator(Unknown Source)
at java.util.stream.ReferencePipeline$Head.forEach(Unknown Source)
at dvp.java8.streams.Exemple02b.main(Exemple02b.java:18)
Ogni volta che si desidera utilizzare un Stream, è necessario generarlo nuovamente, anche se è già stato generato in precedenza.
5.2. Esempio-02 - elaborazione in parallelo degli elementi di uno Stream
 |
Consideriamo il codice seguente:
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple02 {
public static void main(String[] args) {
// elenco delle persone
List<Personne> personnes = Personnes.get();
// visualizzazione 1
personnes.stream().forEach(Exemple02::affiche);
System.out.println("-----------------");
// visualizzazione 2
personnes.stream().parallel().forEach(Exemple02::affiche);
}
public static void affiche(Personne p) {
System.out.printf("Personne %s sur thread %s%n", p, Thread.currentThread().getName());
}
}
- righe 19-21: il metodo [affiche] scrive sulla console la stringa jSON relativa a una persona, nonché il nome del thread di esecuzione in cui avviene la visualizzazione;
- riga 13: visualizza un elenco di persone. Si noti che il parametro del metodo [forEach] è il riferimento al precedente metodo statico;
- riga 16: si esegue la stessa operazione, ma con il metodo [parallel] si richiede che l’elaborazione degli elementi dello stream avvenga in parallelo su più thread. Non tutte le elaborazioni possono essere eseguite in parallelo. In questo caso, si deve presumere che l’ordine di visualizzazione non abbia importanza, poiché in un’elaborazione parallela non è garantito l’ordine di esecuzione dei thread. Si noti inoltre una sintassi che diventerà onnipresente sia per i metodi Stream che per i metodi Observable:
- (continua)
- flux genera elementi e1 che alimentano il metodo m1;
- flux.m1 è a sua volta un flusso di elementi e2 che alimentano il metodo m2;
- flux.m1.m2 è un flusso di elementi e3 che alimentano il metodo m3;
Il tipo degli elementi e1, e2, e3 può variare nel corso delle elaborazioni a cui viene sottoposto il flusso iniziale.
L'esecuzione di questo codice produce il seguente risultato:
Si nota che l’esecuzione parallela (righe 5-7) è avvenuta su tre thread diversi e non ha rispettato l’ordine degli elementi, che è quello delle righe 1-3. In questo documento non ci soffermeremo molto sull’elaborazione parallela degli elementi di un Stream, poiché in tal caso occorrerebbe discutere delle condizioni che rendono possibile tale elaborazione. Si scopre quindi che poche operazioni possono essere eseguite in parallelo. Una di quelle che si presta naturalmente al parallelismo è la somma degli elementi numerici di un flusso, che presentiamo ora.
5.3. Esempio-03 - elaborazione in parallelo degli elementi di uno Stream
 |
Consideriamo il seguente codice (Esempio 03a):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
public class Exemple03a {
public static void main(String[] args) {
final long limite = 10_000_000L;
// numero di processori
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// elenco di numeri
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(i);
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// somma dei numeri - metodo sequenziale
début = new Date().getTime();
long somme = nombres.stream().reduce(0L, (s, i) -> s + i);
System.out.printf("somme séquentielle : somme=%s, durée (ms)=%s%n", somme, new Date().getTime() - début);
}
}
- alla riga 22, utilizziamo il metodo [reduce], la cui firma è la seguente:
 |  |
- il metodo [reduce] opera con elementi di tipo T;
- il metodo [reduce] applica lo stesso trattamento a tutti gli elementi di un flusso: il valore iniziale di un accumulatore viene fornito come primo parametro. Come secondo parametro viene fornito un metodo che istanzia l'interfaccia funzionale [BinaryOperator] [2]: a partire da ciascun elemento e dall'accumulatore, questo metodo fornisce un nuovo valore dell'accumulatore. Il valore finale di quest’ultimo è il valore restituito dal metodo [reduce]. Il codice [3] illustra questo meccanismo. Il metodo [apply] è il metodo dell’interfaccia funzionale [BinaryOperator] [2];
Torniamo al codice di esempio:
- riga 12: viene visualizzato il numero di core rilevati da JVM;
- righe 15-18: si crea un elenco di 10 milioni di numeri;
- riga 22: la somma di questi numeri viene calcolata in modo sequenziale con un unico thread;
Si ottengono i seguenti risultati:
Ora sostituiamo la riga 22 del codice con la seguente (Esempio03b):
long somme = nombres.stream().parallel().reduce(0L, (s, i) -> s + i);
Si richiede che gli elementi dello Stream vengano elaborati in parallelo utilizzando più thread. Ciò è possibile poiché l’ordine di somma dei numeri non ha importanza. È quindi possibile assegnare n1 numeri a un thread T1, n2 numeri a un thread T2, ... e infine sommare i risultati forniti da questi diversi thread. Si ottengono così i seguenti risultati:
Non si riscontra quindi praticamente alcun miglioramento delle prestazioni. Negli esempi che seguiranno sarà spesso così. La gestione dei thread è di per sé dispendiosa in termini di tempo. È necessario che l’operazione eseguita da ciascun core sia sufficientemente complessa affinché si manifesti un miglioramento delle prestazioni. È quanto mostra il seguente esempio (Esempio03c):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.function.BinaryOperator;
public class Exemple03c {
public static void main(String[] args) {
final long limite = 10_000L;
// numero di processori
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// elenco di numeri
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(i);
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// somma dei numeri - metodo sequenziale
début = new Date().getTime();
BinaryOperator<Long> bo = (s, i) -> {
try {
Thread.sleep(1);
} catch (InterruptedException e) {
}
return s + i;
};
long somme = nombres.stream().reduce(0L, bo);
System.out.printf("somme séquentielle : somme=%s, durée (ms)=%s%n", somme, new Date().getTime() - début);
}
}
- riga 30: si utilizza nuovamente il metodo [reduce], al quale viene fornito come parametro il riferimento al metodo delle righe 23-29;
- riga 28: il metodo [bo] restituisce la somma dei suoi due parametri;
- righe 24-27: artificialmente, si fa attendere il thread di 1 millisecondo per simulare un carico di lavoro intenso;
Si ottengono quindi i seguenti risultati:
Ora, se si sostituisce la riga 30 con la seguente:
long somme = nombres.stream().parallel().reduce(0L, bo);
si ottengono i seguenti risultati:
Si nota chiaramente il miglioramento delle prestazioni ottenuto grazie all'esecuzione in parallelo del calcolo della somma. Per l'elaborazione di 8 numeri:
- il thread sequenziale attende 8 volte 1 millisecondo, quindi 8 ms;
- gli 8 thread paralleli attendono contemporaneamente ciascuno 1 millisecondo (per semplificare), quindi in totale 1 millisecondo per gli 8 numeri;
Ci si può quindi aspettare che l’esecuzione parallela sia 8 volte più veloce di quella sequenziale. È più o meno quello che succede in questo caso.
5.4. Esempio-04 - filtrare uno Stream
 |
Consideriamo il seguente codice:
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple04 {
public static void main(String[] args) {
// elenco di persone
List<Personne> personnes = Personnes.get();
// visualizzazioni
System.out.println("age < 28 ----------------------");
personnes.stream().filter(p -> p.getAge() < 28).forEach(p -> {
System.out.println(p);
});
System.out.println("poids < 50 ----------------------");
personnes.stream().filter(p -> p.getPoids() < 50).forEach(p -> {
System.out.println(p);
});
System.out.println("age < 28 ----------------------");
personnes.stream().filter(p -> p.getAge() < 28).forEach(System.out::println);
System.out.println("poids < 50 ----------------------");
personnes.stream().filter(p -> p.getPoids() < 50).forEach(System.out::println);
}
}
- riga 14: il metodo [Stream.filter] ha la seguente firma:
 |
- il metodo [filter] richiede come parametro un'istanza dell'interfaccia funzionale [Predicate] presentata al paragrafo 4.2, il cui unico metodo da implementare è il seguente: boolean test(T t);
- il metodo [filter] restituisce gli elementi dello Stream che soddisfano il Predicate. Serve quindi a filtrare il Stream;
Consideriamo il seguente codice:
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple04 {
public static void main(String[] args) {
// elenco di persone
List<Personne> personnes = Personnes.get();
// visualizzazioni
System.out.println("age < 28 ----------------------");
personnes.stream().filter(p -> p.getAge() < 28).forEach(p -> {
System.out.println(p);
});
System.out.println("poids < 50 ----------------------");
personnes.stream().filter(p -> p.getPoids() < 50).forEach(p -> {
System.out.println(p);
});
System.out.println("age < 28 ----------------------");
personnes.stream().filter(p -> p.getAge() < 28).forEach(System.out::println);
System.out.println("poids < 50 ----------------------");
personnes.stream().filter(p -> p.getPoids() < 50).forEach(System.out::println);
}
}
- righe 14-16: visualizzano le persone di età <28;
- righe 18-20: visualizzano le persone con peso <50;
- riga 22: fa la stessa cosa delle righe 14-16 ma in modo più conciso;
- riga 24: fa la stessa cosa delle righe 18-20, ma in modo più conciso;
I risultati dell'esecuzione sono i seguenti:
5.5. Esempio-05 - Creazione di uno Stream<T2> a partire da uno Stream<T1>
 |
Consideriamo il seguente codice:
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple05 {
public static void main(String[] args) {
// elenco di persone
List<Personne> personnes = Personnes.get();
// visualizzazioni
System.out.println("Personne --> String ----------------------");
personnes.stream().map(p -> p.getNom()).forEach(System.out::println);
System.out.println("Personne --> Integer ----------------------");
personnes.stream().map(p -> p.getAge()).forEach(System.out::println);
}
}
- alla riga 13, il metodo [Stream.map] ha la seguente firma:
 |
Il parametro del metodo [Stream.map] è un'istanza dell'interfaccia funzionale [Function] presentata al paragrafo 4.3, il cui unico metodo da implementare è: R apply(T t). Si nota che, a partire da un tipo T, la funzione [apply] produce un tipo R. Il metodo [Stream.map] produrrà quindi un flusso Stream di tipo R a partire da un flusso di tipo T (per «flusso di tipo T» si intende qui, con un abuso di linguaggio che manterremo, un flusso di elementi di tipo T).
Esaminiamo ora il codice dell’esempio:
- riga 14: di una persona p, si conserva solo il nome. Si ottiene quindi un flusso di tipo String;
- riga 14: di una persona p si conserva solo il nome. Si ottiene quindi un flusso di Integer;
I risultati ottenuti sono i seguenti:
5.6. Esempio-06 - altri metodi della classe Stream<T>
 |
Illustriamo alcuni dei 39 metodi della classe Stream con il seguente codice:
package dvp.java8.streams;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import dvp.data.Personne;
import dvp.data.Personnes;
import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.DoubleStream;
import java.util.stream.IntStream;
import java.util.stream.Stream;
public class Exemple06 {
// mapper jSON
static private ObjectMapper jsonMapper = new ObjectMapper();
public static void main(String[] args) throws JsonProcessingException {
// elenco delle persone
List<Personne> personnes = Personnes.get();
// tutte le persone
affiche("all", personnes);
// la prima persona
affiche("findFirst", personnes.stream().findFirst().get());
// una persona qualsiasi
affiche("findAny", personnes.stream().findAny().get());
// le persone tranne la prima
affiche("skip 1", personnes.stream().skip(1L).collect(Collectors.toList()));
// le prime 2 persone
affiche("limit 2", personnes.stream().limit(2L).collect(Collectors.toList()));
// il numero di persone
affiche("count", personnes.stream().count());
// la persona più anziana
affiche("age max", personnes.stream().max(Comparator.comparingInt(Personne::getAge)).get());
// la persona più leggera
affiche("poids min", personnes.stream().min(Comparator.comparingDouble(Personne::getPoids)).get());
// l'ultima persona in ordine alfabetico dei nomi
affiche("nom max", personnes.stream().max((p1, p2) -> p1.getNom().compareToIgnoreCase(p2.getNom())).get());
// l'età totale di tutte le persone
affiche("âge total (reduce)", personnes.stream().map(p -> p.getAge()).reduce(0, (a1, a2) -> a1 + a2));
// le persone in ordine crescente di età
affiche("personnes par âge croissant",
personnes.stream().sorted(Comparator.comparingInt(Personne::getAge)).collect(Collectors.toList()));
// Ci sono persone che hanno più di 100 anni?
affiche("des personnes de + de 100 ans (anyMatch)", personnes.stream().anyMatch(p -> p.getAge() > 100));
// tutte le persone hanno al massimo 100 anni?
affiche("des personnes de + de 100 ans (noneMatch)", personnes.stream().noneMatch(p -> p.getAge() > 100));
// Tutte le persone hanno più di 8 anni?
affiche("des personnes de + de 8 ans (allMatch)", personnes.stream().allMatch(p -> p.getAge() > 8));
// Le persone vengono raggruppate per sesso
affiche("personnes regroupées par sexe", personnes.stream().collect(Collectors.groupingBy(p -> p.getSexe())));
// eliminazione degli elementi duplicati da una lista
affiche("distinct", Stream.of(1, 2, 1).distinct().collect(Collectors.toList()));
// da uno Stream<Stream<T>> si ottiene uno Stream<T>
affiche("flatMap", Stream.of(1, 2, 3).flatMap(i -> Stream.of(i, i + 10)).collect(Collectors.toList()));
// da uno Stream<Stream<Integer>> si ottiene un IntStream di cui si calcola la somma
affiche("flatMapToInt", Stream.of(1, 2, 3).flatMapToInt(i -> IntStream.of(i, i + 10)).sum());
// di uno Stream<Stream<Integer>>, si crea un DoubleStream e poi un array
affiche("flatMapToDouble", Stream.of(1, 2, 3).flatMapToDouble(i -> DoubleStream.of(i, i * 1.2)).toArray());
// max di un flusso di interi
affiche("reduce Integer::max", Stream.of(1, 10, 8).reduce(Integer::max).get());
// min di un flusso di Double
affiche("reduce Integer::min", Stream.of(1.5, 10.4, 8.9).reduce(Double::min).get());
// media di un flusso di interi
affiche("IntStream average", IntStream.of(1, 10, 8).average().getAsDouble());
// statistiche di un flusso di numeri interi
affiche("IntStream summaryStatistics", IntStream.of(1, 10, 8).summaryStatistics());
}
public static <T> void affiche(String message, T value) throws JsonProcessingException {
System.out.println(String.format("%s ----", message));
System.out.println(jsonMapper.writeValueAsString(value));
}
}
- righe 72, 75: visualizzano la stringa jSON del secondo parametro del metodo;
- riga 24: visualizza la stringa jSON relativa a tutte le persone. Si ottiene il seguente risultato:
5.6.1. [findFirst]
// la prima persona
affiche("findFirst", personnes.stream().findFirst().get());
Il metodo [findFirst] restituisce il primo elemento di un flusso, se presente. La sua firma è la seguente:
 |
Il risultato è di tipo Optional<T>, un tipo introdotto da Java 8:
 |
La classe Optional<T> consente di gestire in modo diverso i puntatori null. Un metodo che deve restituire un tipo T che può assumere il valore null può decidere di restituire un tipo Optional<T>. Il metodo [Optional<T>.isPresent()] consente di verificare se il metodo ha restituito un valore o meno. Il codice seguente [Exemple06b] illustra parte del funzionamento di Optional<T>:
package dvp.java8.streams;
import java.util.Optional;
import com.fasterxml.jackson.core.JsonProcessingException;
public class Exemple06b {
public static void main(String[] args) throws JsonProcessingException {
// opzionale senza valore
Optional<Integer> o1 = m1();
System.out.println(o1.isPresent());
affiche(o1);
// opzionale con valore
Optional<Integer> o2 = m2();
System.out.println(o2.isPresent());
affiche(o2);
}
private static void affiche(Optional<Integer> o1) {
try {
// si recupera il valore dell'Optional
// genera un'eccezione se non c'è alcun valore
System.out.println(o1.get());
} catch (Throwable th) {
System.out.printf("%s : %s%n", th.getClass().getName(), th.getMessage());
}
}
public static Optional<Integer> m1() {
// nessun valore
return Optional.empty();
}
public static Optional<Integer> m2() {
// un valore
return Optional.of(10);
}
}
I risultati ottenuti sono i seguenti:
false
java.util.NoSuchElementException : No value present
true
10
Torniamo al codice illustrativo del metodo [findFirst]:
// la prima persona
affiche("findFirst", personnes.stream().findFirst().get());
- riga 2: per semplificare il codice, utilizziamo il metodo [get] sull'oggetto Optional<Personne> generato dal metodo [findFirst]. Un codice corretto richiederebbe di chiamare il metodo [Optional<Personne>.isPresent()] prima di chiamare il metodo [get];
Il risultato ottenuto è il seguente:
5.6.2. [findAny]
// qualsiasi persona
affiche("findAny", personnes.stream().findAny().get());
Il metodo [findAny] ha la seguente firma:
 |
Il metodo [findAny] può restituire qualsiasi elemento del flusso. Durante i test, si nota che un'esecuzione sequenziale restituisce il primo elemento del flusso, mentre un'esecuzione parallela può effettivamente restituire qualsiasi elemento. Ciò è dimostrato dal seguente codice [Exemple06c]:
package dvp.java8.streams;
import java.util.List;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple06c {
// mappatore jSON
static private ObjectMapper jsonMapper = new ObjectMapper();
public static void main(String[] args) throws JsonProcessingException {
// elenco di persone
List<Personne> personnes = Personnes.get();
// tutte le persone
affiche("all", personnes);
// qualsiasi persona
affiche("findAny parallèle", personnes.stream().parallel().findAny().get());
// qualsiasi persona
affiche("findAny séquentiel", personnes.stream().findAny().get());
}
public static <T> void affiche(String message, T value) throws JsonProcessingException {
System.out.println(String.format("%s ----", message));
System.out.println(jsonMapper.writeValueAsString(value));
}
}
- riga 22: findAny eseguito in parallelo;
- riga 24: findAny eseguito in modo sequenziale;
I risultati ottenuti sono i seguenti:
- riga 4: l'esecuzione parallela ha restituito l'elemento 2 dell'elenco delle persone. Avrebbe potuto essere un altro;
- riga 6: l'esecuzione sequenziale ha restituito il primo elemento dell'elenco delle persone;
L'uso del metodo [findAny] sembra avere senso solo nell'elaborazione parallela di un flusso.
5.6.3. [skip]
// le persone tranne la prima
affiche("skip 1", personnes.stream().skip(1L).collect(Collectors.toList()));
Il metodo [skip] ha la seguente firma:
 |
Il metodo [skip] ignora i primi n elementi di un flusso. Come indicato nella documentazione sopra riportata, l'esecuzione di questo metodo in parallelo comporta scarsi miglioramenti in termini di prestazioni e può addirittura causarne una perdita. Infatti, per ignorare i primi n elementi, i thread sono costretti a coordinarsi, il che annulla i vantaggi in termini di prestazioni derivanti dal parallelismo.
Il metodo [skip] restituisce un flusso Stream<Personne> che viene trasformato in un tipo List<Personne> dal metodo [collect], la cui firma è la seguente:
 |
Il metodo [collect] accetta come parametro un'istanza del tipo [Collector], la cui firma è complessa. Esistono implementazioni predefinite del tipo [Collector] che, nella maggior parte dei casi, consentono di evitare di implementarlo autonomamente. In questo caso l’implementazione utilizzata è [Collectors.toList()]. [Collectors] è una classe che possiede numerosi metodi statici che implementano il tipo [Collector<T,A,R>]. È proprio lì che bisogna cercare quando si vuole trasformare un Stream in una collezione standard di Java:
 |
Utilizzeremo alcuni di questi metodi in seguito.
L’esecuzione produce il seguente risultato:
Il primo elemento dell'elenco (jean) è stato omesso.
5.6.4. [limit]
// le prime 2 persone
affiche("limit 2", personnes.stream().limit(2L).collect(Collectors.toList()));
Il metodo [limit] ha la seguente firma:
 |
Il metodo [limit] consente di conservare solo i primi n elementi di un flusso. Non è adatto all'elaborazione parallela.
L'esecuzione produce il seguente risultato:
5.6.5. [count]
// il numero di persone
affiche("count", personnes.stream().count());
Il metodo [count] ha la seguente firma:
 |
Il metodo [count] restituisce il numero di elementi di un Stream. L'esecuzione in parallelo del metodo non comporta alcun miglioramento delle prestazioni, come mostra il codice seguente (Esempio06d1):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.stream.Stream;
public class Exemple06d1 {
public static void main(String[] args) {
final long limite = 10_000_000L;
// numero di processori
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// elenco di numeri
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(i);
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// conteggio dei numeri - metodo sequenziale
Stream<Long> sNombres = nombres.stream();
début = new Date().getTime();
long count = sNombres.count();
System.out.printf("comptage séquentiel : compteur=%s, durée (ms)=%s%n", count, new Date().getTime() - début);
}
}
- righe 11-22: si crea un Stream con 10 milioni di numeri;
- righe 22-24: conteggio del Stream;
L'esecuzione fornisce il seguente risultato:
Se si sostituisce la riga 22 del codice con la seguente (Esempio06d2):
Stream<Long> sNombres = nombres.stream().parallel();
si ottengono i seguenti risultati:
5.6.6. [max, min]
// la persona più anziana
affiche("age max", personnes.stream().max(Comparator.comparingInt(Personne::getAge)).get());
Il metodo [max] ha la seguente firma:
 |
Il metodo [max] restituisce il valore massimo di un flusso utilizzando il comparatore che gli viene passato come parametro. Comparator è un'interfaccia funzionale il cui unico metodo da implementare ha la seguente firma: int compare (T o1, T o2). Questo metodo deve restituire -1 se o1 < o2, 0 se o1.equals(o2), +1 se o1 > o2. L’interfaccia funzionale Comparator dispone di numerosi metodi statici predefiniti che implementano l’interfaccia Comparator per i casi più comuni. Pertanto, nell’istruzione:
affiche("age max", personnes.stream().max(Comparator.comparingInt(Personne::getAge)).get());
utilizziamo il metodo statico [Comparator.comparingInt], la cui firma è la seguente:
 |
Il tipo ToIntFunction è un'interfaccia funzionale:
 |
Il metodo [applyAsInt] dell'interfaccia funzionale ToIntFunction genera un tipo int a partire da un tipo T. Torniamo al nostro codice:
affiche("age max", personnes.stream().max(Comparator.comparingInt(Personne::getAge)).get());
Il parametro effettivo del metodo [Comparator.comparingInt] deve essere in questo caso un lambda Personne --> int. Passiamo il riferimento al metodo [Personne.getAge], che presenta proprio questa firma. Alla fine, otterremo la persona con l’età più avanzata. Si ottiene un tipo Optional<Personne> dal quale si estrae il valore con il metodo [Optional.get]. Si ottiene il seguente risultato:
Il calcolo parallelo di max non comporta alcun miglioramento delle prestazioni, come mostra il seguente esempio: (Esempio06e1):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.stream.Stream;
public class Exemple06e1 {
public static void main(String[] args) {
// data
// final long limite = 100L;
// final boolean verbose = true;
final long limite = 10_000_000L;
final boolean verbose = false;
// numero di processori
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// elenco di numeri
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(new Random().nextLong());
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// valore massimo dei numeri - metodo sequenziale
Stream<Long> sNombres = nombres.stream();
Comparator<Long> compLong = (l1, l2) -> {
if (verbose) {
// thread
System.out.printf("[%s]", Thread.currentThread().getName());
}
// confronto
long v1 = l1.longValue();
long v2 = l2.longValue();
if (v1 < v2) {
return -1;
} else {
if (v1 == v2) {
return 0;
} else {
return +1;
}
}
};
début = new Date().getTime();
// lunghezza massima = sNombres.max(Comparator.naturalOrder()).get();
long max = sNombres.max(compLong).get();
System.out.printf("%nmax séquentiel : max=%s, durée (ms)=%s%n", max, new Date().getTime() - début);
}
}
- riga 29: si ha un flusso di numeri casuali di tipo Long;
- righe 30-47: la variabile lambda compLong implementa l'interfaccia Comparator<Long>. Questa interfaccia è normalmente implementata dal metodo [Comparator.naturalOrder()] alla riga 49. Ma in questo caso vogliamo visualizzare il thread di esecuzione (righe 31-33). Pertanto implementiamo noi stessi l'interfaccia;
- riga 50: ricerca di max;
Si ottengono i seguenti risultati:
 |
Se ora sostituiamo la riga 27 con la seguente (Esempio06e2):
Stream<Long> sNombres = nombres.stream().parallel();
si ottengono i seguenti risultati:
 |
L'esecuzione parallela è stata quindi più lenta. Se si passa a 10 milioni di numeri con verbose=false, si ottengono i seguenti risultati:
per l'esecuzione sequenziale:
per l'esecuzione parallela, che rimane quindi più lenta.
Si utilizza il metodo [Stream.min] in modo analogo:
// la persona più leggera
affiche("poids min", personnes.stream().min(Comparator.comparingDouble(Personne::getPoids)).get());
5.6.7. [reduce]
// l'età complessiva di tutte le persone
affiche("âge total (reduce)", personnes.stream().map(p -> p.getAge()).reduce(0, (a1, a2) -> a1 + a2));
Il metodo [reduce] è stato presentato al paragrafo 5.3. La riga 2 sopra riportata somma le età di tutte le persone. Il risultato è il seguente:
5.6.8. [sorted]
// le persone in ordine crescente di età
affiche("personnes par âge croissant",
personnes.stream().sorted(Comparator.comparingInt(Personne::getAge)).collect(Collectors.toList()));
// le persone in ordine alfabetico per cognome
List<Personne> lPersonnes=personnes.stream().sorted((p1, p2) -> p1.getNom().compareTo(p2.getNom())).collect(Collectors.toList());
affiche("personnes par ordre alphabétique des noms", lPersonnes);
Il metodo [sorted] (righe 3 e 5) ha la seguente firma:
 |
Il metodo [sorted] accetta come parametro il tipo [Comparator] descritto al paragrafo 5.6.6 per i metodi min e max. Consente di ordinare un Stream secondo l’ordine del comparatore che gli viene passato come parametro. Abbiamo visto che l’interfaccia [Comparator] offre diversi metodi statici predefiniti che implementano i comparatori più comuni, in particolare per numeri e stringhe di caratteri. In questo caso, utilizziamo il metodo [Comparator.comparingInt] che accetta come parametro un tipo ToIntFunction, ovvero un’interfaccia funzionale del metodo [applyAsInt] con la seguente firma: int applyAsInt(T t). Qui, il parametro effettivo passato al metodo [Comparator.comparingInt] alla riga 3 è il riferimento al metodo [Personne.age] che restituisce l’età della persona.
L'interfaccia [Comparator] non offre metodi statici per confrontare stringhe di caratteri. Alla riga 5, costruiamo noi stessi un lambda che implementa l'unico metodo di questa interfaccia: int compare(T t1, T t2)
(p1, p2) -> p1.getNom().compareTo(p2.getNom())
Questa lambda confronta i nomi delle persone. I risultati ottenuti sono i seguenti:
L’esecuzione in parallelo dell’ordinamento non sembra possibile, come mostra il codice seguente (Esempio06f1):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
public class Exemple06f1 {
// mapper jSON
static ObjectMapper jsonMapper = new ObjectMapper();
public static void main(String[] args) throws JsonProcessingException {
// dati
final long limite = 100L;
final boolean verbose = true;
// limite finale lungo = 10_000_000L;
// final boolean verbose = false;
// numero di processori
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// elenco di numeri
long début = new Date().getTime();
List<Integer> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(new Random().nextInt(1000));
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// ordinamento dei numeri - metodo sequenziale
Stream<Integer> sNombres = nombres.stream();
début = new Date().getTime();
Comparator<Integer> compInt = (i1, i2) -> {
if (verbose) {
// thread
System.out.printf("[%s]", Thread.currentThread().getName());
}
// confronto
int v1 = i1.intValue();
int v2 = i2.intValue();
if (v1 < v2) {
return +1;
} else {
if (v1 == v2) {
return 0;
} else {
return -1;
}
}
};
if (verbose) {
affiche("nombres", sNombres.sorted(compInt).collect(Collectors.toList()));
}
System.out.printf("tri séquentiel : durée (ms)=%s%n", new Date().getTime() - début);
}
public static <T> void affiche(String message, T value) throws JsonProcessingException {
System.out.println(String.format("%s ----", message));
System.out.println(jsonMapper.writeValueAsString(value));
}
}
- righe 30-36: si genera un flusso di numeri casuali limite;
- riga 32: si passa la funzione lambda compInt (righe 38-55) al metodo [sorted]. Questa funzione lambda ordina i numeri in ordine decrescente e visualizza il thread che la esegue.
I risultati ottenuti sono i seguenti:
 |
Se nel codice precedente si sostituisce la riga 36 con la seguente (Esempio06f2):
Stream<Integer> sNombres = nombres.stream().parallel();
si ottengono i seguenti risultati:
 |
Si nota che, sorprendentemente, l'ordinamento del flusso di numeri è stato effettuato con un unico thread. Non c'è stato alcun parallelismo. Oppure mi sfugge qualcosa?
5.6.9. [anyMatch, noneMatch, allMatch]
// Ci sono persone che hanno più di 100 anni?
affiche("des personnes de + de 100 ans (anyMatch)", personnes.stream().anyMatch(p -> p.getAge() > 100));
// Tutte le persone hanno al massimo 100 anni?
affiche("des personnes de + de 100 ans (noneMatch)", personnes.stream().noneMatch(p -> p.getAge() > 100));
// Tutte le persone hanno più di 8 anni?
affiche("des personnes de + de 8 ans (allMatch)", personnes.stream().allMatch(p -> p.getAge() > 8));
Alle righe 2, 4 e 6, i metodi [anyMatch, noneMatch, allMatch] hanno come parametro un tipo Predicate descritto al paragrafo 4.2. Essi effettuano quindi un filtraggio. Tutti e tre restituiscono un valore booleano:
- anyMatch restituisce true se esiste almeno un elemento del Stream che soddisfa il filtro;
- noneMatch restituisce true se non esiste alcun elemento di Stream che soddisfi il filtro;
- allMatch restituisce true se tutti gli elementi di Stream soddisfano il filtro;
I risultati ottenuti sono i seguenti:
5.6.10. [collect(Collectors.groupingBy)]
// Le persone vengono raggruppate per sesso
affiche("personnes regroupées par sexe", personnes.stream().collect(Collectors.groupingBy(p -> p.getSexe())));
Il metodo [collect] è stato presentato nel paragrafo 5.6.3. Il suo parametro è un'implementazione dell'interfaccia [Collector]. La classe [Collectors] offre una serie di metodi statici che implementano l’interfaccia [Collector]. Finora abbiamo utilizzato il metodo [Collectors.toList()]. In questo caso utilizziamo il metodo statico [Collectors.groupingBy], che crea un dizionario a partire da Stream. La sua firma è la seguente:
 |
Il metodo [groupingBy] crea, a partire da un tipo Stream<T>, un tipo Map<K,List<T>>. La chiave K è fornita dal parametro del metodo [groupingBy] di tipo Function<T,K>, il cui unico metodo ha la firma: K apply(T t). Se si desidera creare un dizionario indicizzato in base al sesso delle persone, è necessario fornire una funzione che generi il sesso a partire da una persona. In questo caso, come parametro effettivo del metodo [groupingBy], passiamo il riferimento al metodo [Personne.getSexe]. I risultati ottenuti sono i seguenti:
Alla riga 2, abbiamo la stringa jSON di un dizionario indicizzato da due chiavi: HOMME e FEMME.
Il calcolo parallelo non comporta alcun miglioramento delle prestazioni, come dimostra il seguente esempio (Esempio06g1):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
public class Exemple06g1 {
// meglio jSON
static ObjectMapper jsonMapper = new ObjectMapper();
public static void main(String[] args) throws JsonProcessingException {
// dati
final long limite = 100L;
final boolean verbose = true;
// limite finale lungo = 10_000_000L;
// final boolean verbose = false;
// numero di processori
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// elenco di numeri
long début = new Date().getTime();
List<Integer> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(new Random().nextInt(1000));
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// raggruppamento dei numeri per centinaia - metodo sequenziale
Stream<Integer> sNombres = nombres.stream();
Function<Integer, Integer> groupByCent = n -> {
if (verbose) {
System.out.printf("[%s]", Thread.currentThread().getName());
}
return n / 100;
};
début = new Date().getTime();
// Map<Integer, List<Integer>> lNombres = sNombres.collect(Collectors.groupingBy(numero -> numero / 100));
Map<Integer, List<Integer>> lNombres = sNombres.collect(Collectors.groupingBy(groupByCent));
System.out.printf("%nregroupement séquentiel : durée (ms)=%s%n", new Date().getTime() - début);
// risultati
if (verbose) {
affiche("nombres regroupés", lNombres);
}
}
public static <T> void affiche(String message, T value) throws JsonProcessingException {
System.out.println(String.format("%s ----", message));
System.out.println(jsonMapper.writeValueAsString(value));
}
}
- righe 23-38: costruzione di un flusso di numeri limite;
- riga 47: i numeri vengono raggruppati per centinaia. Si utilizza la funzione lambda delle righe 39-44 per poter visualizzare il thread di esecuzione;
I risultati dell'esecuzione sono i seguenti:
 |
Se nel codice si sostituisce la riga 38 con la seguente (Esempio06g2):
Stream<Integer> sNombres = nombres.stream().parallel();
si ottengono i seguenti risultati:
 |
Si nota che l'esecuzione parallela del raggruppamento ha compromesso le prestazioni.
5.6.11. [distinct]
// eliminazione degli elementi duplicati da una lista
affiche("distinct", Stream.of(1, 2, 1).distinct().collect(Collectors.toList()));
Il metodo [distinct] ha la seguente firma:
 |
Consente di eliminare i duplicati da un flusso. Il metodo [Stream.of] (riga 2) ha la seguente firma:
 |
Consente di creare un Stream a partire da valori forniti esplicitamente. I risultati dell’esecuzione sono i seguenti:
5.6.12. [flatMap]
// da uno Stream<Stream<T>>, si ottiene uno Stream<T>
affiche("flatMap", Stream.of(1, 2, 3).flatMap(i -> Stream.of(i, i + 10)).collect(Collectors.toList()));
Il metodo [flatMap] ha la seguente firma:
  |
Il metodo [flatMap] accetta come parametro una funzione che:
- accetta come parametro un elemento di tipo T del Stream;
- restituisce come risultato un flusso Stream<R>;
Se invece del metodo [flatMap], si fosse utilizzato il metodo [map] descritto al paragrafo 5.5, il risultato sarebbe un tipo Stream<Stream<R>> in cui ogni elemento di tipo T del flusso iniziale avrebbe dato origine a un elemento Stream<R>. Il metodo [flatMap] produce invece un tipo Stream<R>. Esso appiattisce (flatten) i diversi flussi Stream<R> in un unico flusso. È quanto mostrano i risultati dell’esecuzione del codice precedente:
Esistono varianti specializzate di [flatMap]:
// da uno Stream<IntStream> si ottiene un IntStream di cui si calcola la somma
affiche("flatMapToInt", Stream.of(1, 2, 3).flatMapToInt(i -> IntStream.of(i, i + 10)).sum());
Il metodo [flatMapToInt] ha la seguente firma:
 |
Il metodo [flatMapToInt] accetta come parametro una funzione che restituisce un tipo IntStream come segue:
 |
IntStream è un flusso derivato da int. Questo tipo è preferibile al tipo Stream<Integer> poiché la sua elaborazione evita operazioni di boxing/unboxing tra i tipi Integer e int. Questa interfaccia riprende numerosi metodi del tipo Stream<T> e ne aggiunge altri, tra cui il metodo [sum] sopra riportato, che somma gli elementi del IntStream.
Il codice seguente illustra l'uso del metodo analogo [flatMapToDouble]:
// da uno Stream<DoubleStream>, si ottiene un DoubleStream e poi un array
affiche("flatMapToDouble", Stream.of(1, 2, 3).flatMapToDouble(i -> DoubleStream.of(i, i * 1.2)).toArray());
Il metodo [DoubleStream.toArray] consente di passare da un tipo DoubleStream a un tipo double[].
I risultati, per questi due esempi, sono i seguenti:
L'esempio seguente mostra i miglioramenti in termini di prestazioni ottenuti passando da un tipo Stream<Long> a un tipo LongStream (Esempio06i1):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
public class Exemple06i1 {
public static void main(String[] args) {
final long limite = 10_000_000L;
// numero di processori
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// elenco di numeri
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(i);
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// somma dei numeri - metodo sequenziale
début = new Date().getTime();
long somme = nombres.stream().reduce(0L, (s, i) -> s + i);
System.out.printf("somme séquentielle du Stream<Integer> : somme=%s, durée (ms)=%s%n", somme, new Date().getTime() - début);
}
}
- riga 22: calcolo della somma di un flusso di numeri di tipo Long;
Si ottengono i seguenti risultati:
Ora sostituiamo la riga 22 con la seguente (Esempio06i2):
long somme = nombres.stream().mapToLong(n -> n.longValue()).sum();
Il metodo Stream<Integer>.mapToLong ci permette di ottenere un flusso di tipo LongStream di elementi di tipo primitivo long, che poi sommiamo con la funzione sum. Si ottengono quindi i seguenti risultati:
Il miglioramento delle prestazioni è evidente.
5.6.13. Metodi di flusso dei numeri primitivi
// valore massimo di un flusso di interi
affiche("IntStream max", IntStream.of(1, 10, 8).max());
// minimo di un flusso di double
affiche("DoubleStream min", DoubleStream.of(1.5, 10.4, 8.9).min());
// media di un flusso di interi
affiche("IntStream average", IntStream.of(1, 10, 8).average().getAsDouble());
// statistiche di un flusso di interi
affiche("IntStream summaryStatistics", IntStream.of(1, 10, 8).summaryStatistics());
I flussi di valori primitivi (int, long, double) offrono metodi adatti a questi tipi. Il risultato dell'esecuzione del codice precedente è il seguente:
- il risultato della riga 2 del codice è un tipo OptionalInt analogo al tipo Optional<Integer>. Il valore memorizzato in questo oggetto può essere ottenuto con il metodo [getAsInt()]. La presenza di un valore può essere verificata con il metodo [isPresent()]. La riga 2 dei risultati non significa che la classe [OptionalInt] abbia campi denominati [asInt, present]. Per impostazione predefinita, la libreria jSON utilizza tutti i metodi pubblici getX e isY dell’oggetto da serializzare in jSON. E in questo caso, esiste effettivamente un metodo [getAsInt] e un altro metodo [isPresent], mentre i campi [asInt, present] non esistono;
- il risultato della riga 4 del codice è un tipo OptionalDouble analogo al tipo Optional<Double>;
- il risultato della riga 6 del codice è un tipo OptionalDouble il cui valore può essere ottenuto con il metodo [getAsDouble()]. Il metodo [average] calcola la media della sequenza di numeri;
- il risultato della riga 8 del codice è un tipo IntSummaryStatistics definito come segue:
 |
Si nota che l’oggetto IntSummaryStatistics ottenuto fornisce diverse informazioni sul flusso di numeri, quali il numero di valori, la somma, il massimo, il minimo e la media.