Nel abbiamo introdotto le lambda expression: funzioni anonime che possiamo utilizzare dove sia prevista un’interfaccia con un solo metodo astratto. In questo articolo useremo le lambda con gli stream, un nuovo strumento che ci permetterà di lavorare con le collezioni in maniera efficiente e compatta grazie proprio alla programmazione funzionale, ma andiamo subito a vedere il codice.

The Division Bell

Procuriamoci una collezione: useremo un ArrayList di Album. Questa classe ha una struttura molto semplice: un titolo, l’autore, l’anno di pubblicazione e la lista di canzoni.

package it.cosenonjaviste.stream;

import java.util.*;

public class Album {
   
   private String author;
   private List songs;
   private int year;
   private String title;
   
   public Album(String author, String title, int year, String... songs){
      this.author = author;
      this.year = year;
      this.title = title;
      this.songs = new ArrayList<>(Arrays.asList(songs));
   }

   public String getAuthor() {
      return author;
   }

   public List getSongs() {
      return songs;
   }

   public int getYear() {
      return year;
   }
   
   public String getTitle() {
      return title;
   }

}

Per capire meglio cosa può contenere un Album, riportiamo le prime righe del codice che abbiamo usato per inizializzare la lista.

      List albums = new ArrayList<>();
      albums.add(new Album("Pink Floyd", "The Division Bell", 1994,
               "Cluster One",
               "What Do You Want from Me",
               "Poles Apart",
               "Marooned",
               "A Great Day for Freedom",
               "Wearing the Inside Out",
               "Take It Back",
               "Coming Back to Life",
               "Keep Talking",
               "Lost for Words",
               "High Hopes"));

Siamo pronti per iniziare. Scriviamo una semplice operazione sui dati: restituire la lista degli autori che hanno pubblicato qualcosa prima del 2000. Confronteremo due versioni: quella in Java 7:

      for (Album album : albums)
         if (album.getYear() < 2000)
            System.out.println(album.getAuthor());

e quella in Java 8 con gli stream.

      albums.stream()
               .filter(album -> album.getYear() < 2000)
               .map(Album::getAuthor)
               .forEach(System.out::println);

C’è una bella differenza tra i due codici scritti. Nel primo caso osserviamo delle istruzioni familiari: prendere un singolo elemento dalla lista, processarlo e avanzare verso il successivo. Conosciamo già questo modo di operare, si chiama iterazione esterna e ne abbiamo parlato a lungo nel post precedente. Nel secondo caso, invece, ci facciamo restituire un oggetto stream dalla collezione. Su questo usiamo il metodo filter che ci permette di considerare solo gli album che soddisfano il nostro criterio tramite una lambda. Sugli album che superano la selezione, tramite map e con un method reference, ci facciamo tornare solo l’autore e andiamo a stampare ciascuno di questi elementi.

Hai detto stream?

Ma cos’è allora uno stream? Nessuna parentela con InputStream o OutputStream. Uno stream è un’interfaccia che restituisce un flusso di dati finito o infinito su cui è possibile fare operazioni di filtro, mappa e riduzione (leggiamo aggregazione per il momento). Può operare in modo sequenziale (che è il default) oppure a richiesta in modo parallelo. Possiamo vedere lo stream come una pipeline di operazioni su dati che, come ogni tubatura che si rispetti, ha una sorgente, una serie di pezzi intermedi e una destinazione.

La sorgente

La prima cosa da chiarire è che uno stream non contiene dati a parte una manciata di flag che consentono di controllare e ottimizzare il flusso. Ha bisogno quindi di una sorgente da cui pescare informazioni. L’esempio più comune è consiste nel partire da una collection, come sopra, ma uno stream può essere generato anche in altri modi come riportato qui sotto.

IntStream intStream = IntStream.range(1, 10);
DoubleStream doubleStream = new Random().doubles();
Stream pathStream = Files.list(Paths.get("c:\\"));

Nella prima riga costruiamo un IntStream a partire da un intervallo di interi. IntStream non è altro che una specializzazione di Stream specifica per gli interi; allo stesso modo Java 8 fornisce DoubleStream e LongStream. Nel secondo esempio abbiamo uno stream infinito di double pseudocasuali compresi tra 0 e 1. Infine, a partire da un percorso su file system (stiamo usando java.nio), restituiamo uno stream di Path.

Le operazioni intermedie

Una pipeline può avere zero o più operazioni intermedie. La caratteristica principale di questi metodi è quella di essere lazy, cioè il dato non viene mai elaborato, l’elaborazione avviene solo nel passo finale. Cosa fa allora un’operazione intermedia? Restituisce un nuovo stream: per rendere meglio questo aspetto, riscriviamo il primo esempio di stream esplicitando i tipi in gioco.

Stream stream1 = albums.stream();
Stream stream2 = stream1.filter(album -> album.getYear() < 2000);
Stream stream3 = stream2.map(Album::getAuthor);
stream3.forEach(System.out::println);

Osserviamo da vicino il metodo filter che è la prima operazione che abbiamo usato:

Stream filter(Predicate super T> predicate); //Returns a stream consisting of the elements of this stream that match the given predicate.

Come conferma la documentazione, questa operazione restituisce uno stream che soddisfa un Predicate passato come parametro. Un Predicate (predicato in italiano) è una interfaccia funzionale (come poteva non esserlo, abbiamo passato una lambda 🙂 ) che ha un unico compito: quello di rispondere vero o falso nel metodo test. Gli oggetti che soddisfano il test andranno avanti nella pipeline, gli altri saranno scartati.

L’altra operazione intermedia del nostro esempio è un mapping, cioè trasformiamo gli oggetti nello stream corrente in altri oggetti. Anche in questo caso val la pena dare un occhio alla firma del metodo:

 Stream map(Function super T, ? extends R> mapper); // Returns a stream consisting of the results of applying the given function to the elements of this stream.

Anche map torna uno stream, però, a differenza di filter, ci permette di cambiare il tipo di oggetti nello stream passando una lambda che sia compatibile con l’interfaccia funzionale Function. Questa interfaccia richiede di implementare il metodo apply che prende un oggetto di tipo T, quello delle stream corrente e torna un oggetto di tipo R. Nell’esempio di sopra, ad uno stream di Album passiamo una lambda che dato un album ritorna il suo titolo.

Come è facile immaginare, ci sono altre operazioni intermedie ma vogliamo seguire il nostro esempio fino in fondo in modo da avere innanzitutto una visione di insieme.

L’operazione terminale

L’ultima operazione dello stream è detta terminale ed è quella che effettivamente agisce sui dati. Cosa possiamo fare al termine dello stream? Possiamo farci dare, ad esempio una somma o una media se abbiamo un IntStream, un DoubleStream o un LongStream. Possiamo attraversare lo stream ed eseguire un’operazione, come nell’esempio di partenza dove utilizziamo un forEach. Possiamo costruire una lista con il risultato della pipeline e su questa effettuare dei raggruppamenti. Vedremo in un prossimo post molto più in dettaglio le operazioni terminali, ma giusto per dare un’anticipazione delle possibilità proponiamo due casi.

List authors = albums.stream()
    .filter(album -> album.getYear() < 2000)
    .map(Album::getAuthor)
    .collect(Collectors.toList());
        
double avgSong = albums.stream()
    .mapToInt(album -> album.getSongs().size())
    .average().getAsDouble();

Nel primo esempio, al posto di applicare un forEach, tramite il metodo collect riversiamo il risultato dello stream in una lista. Nel secondo, torniamo per ogni album il numero di canzoni e ne calcoliamo la media (come double). Notiamo una cosa: il metodo average non è disponibile per qualsiasi stream, ma solo per quelli numerici, per cui non possiamo usare semplicemente map ma dobbiamo convertire lo stream in un IntStream tramite mapToInt.

Effetto farfalla

Abbiamo esaminato in dettaglio il codice all’inizio del post, riconfrontiamolo con la soluzione pre-Java-8. Quello che appare più evidente è che lavoriamo dicendo allo stream cosa fare, non come farlo, di conseguenza ciò che abbiamo scritto è più simile al problema di partenza, piuttosto che a una procedura per risolverlo.

Facciamo l’esperimento inverso, che fa questo codice?

int limit = 10;
List songs = new ArrayList<>();
for (Album album : albums) {
   if (album.getYear() < 2000) {
      songs.addAll(album.getSongs());
   }
}
Collections.sort(songs);
for (int i = 0; i < limit; i++) {
   System.out.println(songs.get(i));
}

Probabilmente, dobbiamo leggerlo un paio di volte prima di essere sicuri del significato: prepara un intero e una lista di stringhe, poi per ogni album, se l’album è pubblicato prima del 2000, aggiunge tutte le canzoni alla lista. Ordina la lista e poi stampa i primi 10 elementi. Detto altrimenti è dammi le prime 10 canzoni in ordine alfabetico pubblicate prima del 2000.

Questo frammento, invece, cosa fa?

albums.stream()
    .filter(album -> album.getYear() < 2000)
    .flatMap(album -> album.getSongs().stream())
    .sorted()
    .limit(10)
    .forEach(System.out::println);

Risposta: esattamente la stessa cosa ma è più semplice da leggere perché non ci sono variabili intermedie, non ci sono cicli e l’aspetto è più simile alla formulazione del problema. Dobbiamo solo chiarire cosa fanno le nuove operazioni che abbiamo introdotto: sorted e limit hanno un significato abbastanza intuitivo: la prima ordina gli elementi dello stream, la seconda limita il numero di elementi. Diverso è invece il discorso per flatMap di cui riportiamo la firma.

 Stream flatMap(Function super T,? extends Stream extends R>> mapper)

Questo metodo prende in ingresso una funzione che va dal tipo T dello stream di partenza ad uno stream di R. Il che vuol dire che è una funzione che trasforma uno stream di oggetti in uno stream di stream di un nuovo tipo. Ma dato che il risultato deve essere uno stream e basta, flatMap “schiaccia” tutti gli oggetti in un unico stream eliminando gli stream intermedi. Nell’esempio abbiamo passato una lambda tale che, dato un album, restituisce lo stream delle canzoni di ciascun album. Il risultato “schiacciato” è quindi l’elenco di tutte le canzoni degli album dello stream di partenza.

Abbiamo detto che gli stream sono “cool” perché il codice si legge come il problema di partenza; il vantaggio non finisce qui. Nel caso in cui il problema cambi anche di poco, senza gli stream c’è il rischio di modificare molte righe, con gli stream, invece, c’è meno codice da riscrivere. Diamone una dimostrazione modificando leggermente il problema. Vogliamo le prime 10 canzoni che iniziano con la lettera “D”, in ordine alfabetico, pubblicate prima del 2000 ma senza ripetizioni.

Prima la soluzione in Java 7: sono evidenziate le linee che hanno subito cambiamenti rispetto alla versione precedente.

int limit = 10;
Set songs = new TreeSet<>();
for (Album album : albums) {
   if (album.getYear() < 2000) {
      for (String song : album.getSongs()) {
         if (song.startsWith("D"))
            songs.add(song);
      }
   }
}
int temp = 0;
for (String s : songs) {
   System.out.println(s);
   temp++;
   if (temp == limit)
      break;
}

La soluzione con gli stream.

albums.stream()
    .filter(album -> album.getYear() < 2000)
    .flatMap(album -> album.getSongs().stream().filter(s -> s.startsWith("D")))
    .distinct()
    .sorted()
    .limit(10)
    .forEach(System.out::println);

Nella versione senza stream richiedere un risultato senza ripetizioni ci ha costretto a cambiare il tipo di songs e, di conseguenza, il codice che lo manipolava. In Java 8 è bastato aggiungere l’operazione intermedia distinct che elimina i duplicati. In entrambe le soluzioni, il filtro sulla lettera iniziale non ha stravolto il codice; notiamo che con gli stream abbiamo sicuramente una soluzione più elegante.

Parallelismo

A questo punto dovremmo essere abbastanza convinti della “resistenza” degli stream alle modifiche del problema, ma, se ci fosse ancora qualche dubbio, aggiungiamo un ultimo requisito: il codice deve girare in parallelo su più core della macchina in modo da ottenere tempi di risposta più veloci. Utilizzando gli stream, fare questo è facile, senza …no 🙂 .

Ad esempio, per contare quante canzoni ci sono nella nostra collezione di album possiamo scrivere:

long count = albums.stream()
                   .flatMap(album -> album.getSongs().stream())
                   .count();

e se vogliamo fare questo in parallelo cambiamo il codice in questo modo.

long count2 = albums.parallelStream()
                    .flatMap(album -> album.getSongs().stream())
                    .count();

Tutto qui? Da un punto di vista sintattico, sì. Tuttavia la portata della novità è notevole. Lo strumento più avanzato e semplice a disposizione per andare in parallelo finora è stato il framework Fork/Join (di cui anche in CoseNonJaviste), tuttavia non è stato ritenuto abbastanza facile da usare. Ora, invece, cambiando lo stream di partenza possiamo chiedere alla libreria di andare in parallelo nascondendoci tutti i dettagli. Buttato fuori dalla porta, Fork/Join rientra dalla finestra, perché andando a richiedere una computazione parallela, lo stream usa Fork/Join al posto nostro semplificandoci la vita.

Quando utilizziamo uno stream parallelo, dobbiamo sempre tenere presente due cose. La prima è che introduciamo il non determinismo. Ad esempio, se cerchiamo in parallelo un elemento nello stream che abbia certe caratteristiche e ce n’è più di uno, potremmo ottenere risposte diverse in esecuzioni diverse. Possiamo avere dei grattacapo anche quando abbiamo del codice che produce dei side-effects come quello qui sotto:

List results = new ArrayList<>();
albums.parallelStream()
      .flatMap(album -> album.getSongs().stream())
      .forEach(album -> results.add(album)); // DON'T DO THIS AT HOME!

Queste righe, in maniera non prevedibile, producono un’eccezione, perchè ArrayList non è thread-safe. In questo caso la soluzione è:

List results2 = albums.parallelStream()
                              .flatMap(album -> album.getSongs().stream())
                              .collect(Collectors.toList());

Il secondo punto di attenzione è che parallelo non vuol dire necessariamente più veloce. Andare in parallelo implica che i dati vengono divisi, elaborati parallelamente e riuniti. Se le fasi di divisione e di merge sono troppo costose, potrebbero annullare i benefici del parallelismo ottenendo paradossalmente un tempo di risposta superiore all’esecuzione sequenziale. Molto può dipendere dalla natura della sorgente dello stream, dalla quantità di dati coinvolti e dall’operazione terminale. E’ per questo che per gli stream il default è sequenziale, ma il programmatore può passare all’esecuzione parallela sapendo che “da grandi poteri derivano grandi responsabilità” 🙂 .

Cosa non fare con uno stream

In uno stream, parallelo o sequenziale, le operazioni nella pipeline non devono modificare la sorgente dati (con alcune eccezioni per le operazioni intermedie) e le lambda che vengono passate devono essere stateless, non devono, cioè, dipendere dallo stato di qualche altro oggetto (anche esterno alla pipeline). Se questi due accorgimenti non vengono rispettati, si può incappare in falsi risultati o eccezioni runtime. Inoltre, uno stream non può essere impiegato più dopo l’esecuzione dell’operazione terminale. Riprendiamo uno degli esempi fatti sopra ed aggiungiamo una riga.

Stream stream1 = albums.stream();
Stream stream2 = stream1.filter(album -> album.getYear() < 2000);
Stream stream3 = stream2.map(Album::getAuthor);
stream3.forEach(System.out::println);

stream2.forEach(System.out::println);

La riga 6 darà IllegalStateException dicendo che lo stream è già stato utilizzato.

Conclusioni

In questo post abbiamo introdotto gli stream e visto come adoperarli per interrogare i dati. Alcuni argomenti, come le operazioni terminali, sono stati solo sfiorati, ma ci saranno altri post sul tema.

I cambiamenti di Java 8 non sono stati limitati “solo” alle lambda: siamo di fronte anche ad un miglioramento delle librerie. Con un modo semplice per passare il comportamento a metodi, e grazie all’inversione di controllo, il confine tra codice client e libreria è più netto. Il client decide cosa fare, mentre la libreria decide come, e come ottimizzare il lavoro.

Abbiamo quindi uno strumento molto potente a disposizione e nell’immediato futuro il paragone con altri framework simili sarà inevitabile.