32

Stream API

• Introdotta in Java 8, la java.util.stream forniscesupporto per operazioni in stile funzionale su un flusso di valori

3333

Stream API – Cheat Sheet

34

Cos’è un Stream?

• Un insieme di oggetti, tipicamente provenienti da una collezione, array, generatore o qualche operazione di I/O, dati che pronti ad essere elaborati

• Non è una collezione (e.g., non si possono memorizzare dati, non ha memoria) ma è un astrazione, una ricetta che definisce un insieme di passi su come i dati verranno poi elaborati. Elaborazione a pipeline composta da

• Un generatore di stream (sorgente, una vista sui dati da elaborare)

• Zero o più operazioni intermedie dette anche lazzy

• Una singola operazione terminale detta anche eager

• Maggiormente usate al supporto di elaborazione in stile funzionale dei dati, non producendo alcun side-effect

• A differenza delle collezioni che sono finite, le stream sono potenzialmente infinite. Gli operatori di short-circuiting come limit(n) or findFirst() permettono computazioni su stream infinite di concludere in tempo finito.

• Consumabile: gli elementi di un Stream sono visitati solo una volta. Come nel caso dell'Iterator, una nuova stream dovrebbe essere generata per ri-visitare i sui elementi

• Fornisce supporto al parallelismo (vedrete in seguito): .stream() vs .parallelStream()

3535

Streams - Laziness

• Una sequenza di elementi provenienti da una sorgente

• Non e’ una struttura dati: non salva elementi

• Si presta all’utilizzo delle pipeline di processamento

• N operazioni intermedie e 1 terminale

• Cosiddette Lazy: solo l’operazione terminale fa si che il calcolo inizii.

List<Employee> employess = ... employees.stream()

.filter(e -> e.getIncome() > 50000)

.map(e -> e.getName())

.forEach(System.out::println);

3636

Operazioni intermedie, terminali

Intermedie

distinct

map, flatMap

limit, skip peek

sorted

Terminali

collect

count

forEach min,

max reduce

toArray

findAny, findFirst

allMatch, anyMatch, noneMatch

Restituisce un’altra stream la qualle puo’ essere sottoposta ad’altre operazioni

Restituisce il risultato finale della pipeline di processamento

37

Operazioni su Stream

• Intermedie: l’output è un altro stream

• Stream<T> filter(Predicate<? super T> predicate): ritorna un stream composta da elementi di questo stream per i qualli il predicato e vero - Stateless

Esempi

List<Integer> list = Stream<Integer>.of(1, 2, 3)

.filter(x -> x.intValue() > 1)

.collect(Collectors.toList());

List<Employee> list = getEmployees();

long employee_number = list.stream().filter(x -> x).count();

list.stream().filter(x -> x.getAge() > 70).count();

38

Operazioni su Stream

• Intermedie: l’output è un altro stream

• Stream<R> map(Function<? super T,? extends R> mapper): ritorna un stream che consiste in elementi di questo stream ai qualli applico la funzione passata in argomento - Stateless

• Esempi

Stream.of(‘’A’’, ‘’Bb’’, ‘’Ccc’’)

.mapToInt(string -> string.length()).max()//String::length

List<Employee> employees = getEmployees();

OptionalInt max_age = employees.stream()

.mapToInt(x -> x.getAge())

.max();

3939

• Intermedie: l’output e’ un altro stream

• <R> Stream<R> flatMap(Function<? super T,? extends

Stream<? extends R>> mapper): produce un stream di stream (lineare),

rimpiazando ogni elemento dello stream di partenzacon un stream, prodotto applicando la funzione passate come argomento - Stateless

Operazioni su Stream

4040

Operazioni su stream

• Esempi

List<Integer> together = Stream.of(Arrays.asList(1, 2, 3), Arrays.asList(4, 5, 6)).flatMap(x -> x.stream()).collect(Collector::toList());

List<Employee> employees = ceo.getDirectors().stream().flatMap(x::getEmployes).collect(Collector.toList())

CEO Director Employee

4141

Operazioni su Stream

• Intermedie: l’output è un altro stream

• Stream<T> distinct(): ritorna un stream di elementi distinti (secondo Object.equals(Object)) applicato agli elementi di questo stream

EsempioList<Integer> list = Stream.of

(Arrays.asList(1,1,2),Arrays.asList(3,3,2)).flatMap(x x.stream()).distinct().collect(Collectors.toList());

• Stream<T> sorted(): ritorna un stream di elementi ordinati secondo il loro ordine naturale. Se gli elementi non sono Comparable, viene lanciata una java.lang.ClassCastException (quando applicata un operazione terminale)

List<Integer> list = Stream.of(Arrays.asList(1,1,2),Arrays.asList(3,3,2)).flatMap(x x.stream()).sorted().collect(Collectors.toList());

42

Operazioni su Stream

• Operatori terminali:

• void forEach(Consumer<? super T> action): compie un azione per ogni element di questo stream e.g., Stream.of(Array.asList(1,2,3)).forEach(System.out:

:println)

• long count(): ritorna il numero di elementi presenti in questo stream e.g., Stream.of(Array.asList(-1, -2, 3)).stream.filter(x x > 0).count()

• boolean allMatch/noneMatch(Predicate<? super T>

predicate): ritorna true se tutti/nessun elemento rende true il predicato e.g., Stream.of(Array.asList(1,2,3).allMatch(x x >

0))

43

Operazioni su Stream

• Operatori terminali:

• T reduce(T identity, BinaryOperator<T> accumulator): ritorna una riduzione degli elemento di questo stream, usando il valore identità e funzione associativa entrambe passate come argomento. Questo operatore terminale è equivalente a:T result = identity;

for (T element : this stream)

result = accumulator.apply(result, element)

return result;

• Esempio

List<Integer> integers = Array.asList(1,2,3,4);

Integer sum = integers.stream().reduce(0, (a, b) -> a+b);

--List<String> list = Stream.of(‘asdas’,’asdsd’,’asdasd’ …);

list.stream().mapToInt(x String::length).reduce(0, (x,y)

x+y);

==

BiOperator<Integer> funct = (x,y) x+y;

list.stream().mapToInt(x String::length).reduce(0, funct);

--

list.stream().mapToInt(x String::length).average().getAsDouble()

4444

reduce : operazione terminale

Stream

reduce(0, (a, b) -> a + b)

Integer

4545

Esempio Pipeline - reduce

List<String> strings = asList("Lambda", "expressions", "are","easy", "and", "useful");

int totalLength = strings.stream().map(String::length).reduce(0, (a, b) -> a + b);

46

Operazioni su Stream

• Operatori terminali: • Optional<T> min/max(Comparator<? super T> comparator): ritorna l'elemento

min/max di questo stream secondo il Comparator passato come argomento.

@FunctionalInterface

public interface Comparator<T> {

int compare(T o1, T o2); //+ a variety of default/static methods

}//impone un ordine totale sui elementi, ritorna < 0, 0, > 0 se

//il primo argomento e minore, uguale, maggiore del secondo

Comparator<String> comparator = (x, y) -> x.length() – y.length();

List<String> list = new Arrays.asList(‘1’, ‘11’, ‘111’…);

Optional<String> value = list.stream().min(comparator);

If(value.isPresent()) System.out.println(value.get());

--

//limit, min, max

IntStream intStream = new Random().ints(100, 1, 10_000);

intStream.max();

47

Esempio #3File come sorgenteStream

4848

.limit(40)

.collect(toList());

Separazione responsabilita’

List<String> errors = new ArrayList<>(); int errorCount = 0;File file = new File(fileName);String line = file.readLine();while (errorCount < 40 && line != null) { if (line.startsWith("ERROR")) {errors.add(line);

errorCount++;}line = file.readLine();}

List<String> errors = Files.lines(Paths.get(fileName).filter(l -> l.startsWith("ERROR")

4949

Raggruppamento / Grouping

Map<Dish.Type, List<Dish>> dishesByType = new HashMap<>();

for (Dish dish : menu) {Dish.Type type = dish.getType(); List<Dish> dishes = dishesByType.get(type); if (dishes == null) {

dishes = new ArrayList<>(); dishesByType.put(type, dishes);

}dishes.add(dish);

}

5050

Collectors - groupingBy

classify item into list

keyapply

item

grouping Map

fish meat other

salmon pizza rice

french

fries

pork beef

chicken

prawns

Classification

Function fish

Stream

next

Map<Dish.Type, List<Dish>> dishesByType = menu.stream()

.collect(groupingBy(Dish::getType));

51

Valutazione delle operazioni

• Le operazioni intermedie (lazzy) non sono valutate fino a quando un operazione terminale viene incontrata

• Ogni operazione intermedia restituisce un oggetto di tipo Stream

• Ad ogni passo si aggiunge un nuovo ‘’ingrediente’’ alla ricetta (pipeline delle operazioni)

• Questa scelta permette di fare

• ottimizzazione di codice durante la compilazione

• evitare di fare buffering di stream intermedi

• gestione più semplice dei stream paralleli

5252

Attenzione

• Le espressioni lambda non devono interferire con la stream.

try {

List<String> listOfStrings =

new ArrayList<>(Arrays.asList("one", "two"));

String concatenata = listOfStrings.stream()

// Interferenza

.filter(s -> listOfStrings.add("three"); return true)

.reduce((a, b) -> a + " " + b)

.get();

System.out.println("Concatenated string: " +

concatenatedString);

} catch (Exception e) {

System.out.println("Exception caught: " + e.toString());

}//ConcurrentModificationException

5353

NullPointerException

Raise your hand Ifyou’ve ever seen this

5454

Null references? No, Thanks

✗

✗

✗

Errors source → NPE is by far the most common exception in Java

Bloatware source → Worsen readability by making necessary to fill our code with null checks

Breaks Java philosophy → Java always hides pointers to developers, except in one case: the null pointer

Tony Hoare, who invented the null reference in 1965 while working on an object oriented language called ALGOL W, called its invention his

“billion dollar mistake”

5555

Replacing nulls with Optionals

value

If nulls are so problematic why don't we just avoid them?

value

EMPTY

Optional

null

Optional is a type that models a possibly missing value

Optional

5656

public class Person { private Car car;public Car getCar() { return car; }

}

public class Car {private Insurance insurance;public Insurance getInsurance() {

return insurance; }

}

public class Insurance { private String name;public String getName() { return name; }

}

Finding Car's Insurance Name

5757

String getCarInsuranceName(Person person) {

if (person != null) {Car car = person.getCar(); if (car != null) {

Insurance insurance = car.getInsurance(); if (insurance != null) {

return insurance.getName()}

}}return "Unknown";

}

Attempt 1: deep doubts

5858

Attempt 2: too many choices

String getCarInsuranceName(Person person) { if (person == null) {

return "Unknown";}Car car = person.getCar(); if (car == null) {

return "Unknown";}Insurance insurance =car.getInsurance(); if (insurance == null) {

return "Unknown";}return insurance.getName()

}

5959

Optional to the rescue

public class Optional<T> {private static final Optional<?> EMPTY = new Optional<>(null); private final T value;

private Optional(T value) { this.value = value;

}

public<U> Optional<U> map(Function<? super T, ? extends U> f) { return value == null ? EMPTY :

new Optional(f.apply(value));}

public<U> Optional<U> flatMap(Function<? super T, Optional<U>> f) { return value == null ? EMPTY : f.apply(value);

}}

6060

public class Car {private Optional<Insurance> insurance;public Optional<Insurance> getInsurance() { return insurance; }}

public class Insurance { private String name;public String getName() { return name; }}

Rethinking our model

public class Person {

private Optional<Car> car;

public Optional<Car> getCar() { return car; }

}

•Using the type systemto model nullable value

6161

String getCarInsuranceName(Optional<Person> person) { return person.flatMap(person -> person.getCar())

.flatMap(car -> car.getInsurance())

.map(insurance -> insurance.getName())

.orElse("Unknown");}

Restoring the sanity

6262

Restoring the sanity

Person

String getCarInsuranceName(Optional<Person> person) { return person.flatMap(person -> person.getCar())

.flatMap(car -> car.getInsurance())

.map(insurance -> insurance.getName())

.orElse("Unknown");}

Optional

6363

Restoring the sanity

String getCarInsuranceName(Optional<Person> person) { return person.flatMap(person -> person.getCar())

.flatMap(car -> car.getInsurance())

.map(insurance -> insurance.getName())

.orElse("Unknown");}

flatMap(person -> person.getCar())Person

Optional

6464

Restoring the sanity

String getCarInsuranceName(Optional<Person> person) { return person.flatMap(person -> person.getCar())

.flatMap(car -> car.getInsurance())

.map(insurance -> insurance.getName())

.orElse("Unknown");}

flatMap(person -> person.getCar())Car

Optional

Optional

6565

Restoring the sanity

String getCarInsuranceName(Optional<Person> person) { return person.flatMap(person -> person.getCar())

.flatMap(car -> car.getInsurance())

.map(insurance -> insurance.getName())

.orElse("Unknown");}

flatMap(car -> car.getInsurance())

car

Optional

6666

Restoring the sanity

String getCarInsuranceName(Optional<Person> person) { return person.flatMap(person -> person.getCar())

.flatMap(car -> car.getInsurance())

.map(insurance -> insurance.getName())

.orElse("Unknown");}

flatMap(car -> car.getInsurance())

Insurance

Optional

Optional

6767

Restoring the sanity

String getCarInsuranceName(Optional<Person> person) { return person.flatMap(person -> person.getCar())

.flatMap(car -> car.getInsurance())

.map(insurance -> insurance.getName())

.orElse("Unknown");}

map(insurance -> insurance.getName())insurance

Optional

6868

Restoring the sanity

String getCarInsuranceName(Optional<Person> person) { return person.flatMap(person -> person.getCar())

.flatMap(car -> car.getInsurance())

.map(insurance -> insurance.getName())

.orElse("Unknown");}

orElse("Unknown")String

Optional

6969

How are Java streams implemented?

• java.util.SplitIterator<T> {boolean tryAdvance(Consumer<? super T> action);

default void forEachRemaining(Consumer<? super T> action) {do { } while (tryAdvance(action));

}

Spliterator<T> trySplit();

long estimateSize();

default long getExactSizeIfKnown() {…}

public static final int ORDERED = 0x00000010;

public static final int DISTINCT = 0x00000001;

public static final int SORTED = 0x00000004;

public static final int SIZED = 0x00000040;

…

}

• An object for traversing and partitioning elements of a source. The source of elements covered by a Spliterator could be, for example, an array, a Collection, an IO channel, or a generator function.

• Many method calls (well inlined/fused by the JIT)

7070

stream()employees. parallelStream().filter(e -> e.getRole() == Role.MANAGER).map(Employee::getIncome).reduce(0, (a, b) -> a + b);

Streams – Parallelism for free

7171

Example of parallel reduction

72

Is there such thing as a free lunch?

7373

Probably yes …

… but we need functional forks and knives to eat it

7474

Concurrency & Parallelism

Parallel programming Running multiple tasks at

the same time

Concurrent programmingManaging concurrent requests

Both are hard!

7575

The cause of the problem …

Mutable state +Parallel

processing =Non-determinism

Functional Programmin

g

7676

Too hard to think about them!

Race conditions

Deadlocks

Starvation

Livelocks

… and its

effects

7777

The native Java concurrency model

Based on:

They are sometimes plain evil …

… and sometimes a necessary pain …

… but always the wrong default

Threads

Semaphores

SynchronizationLocks

7878

Different concurrency models

Isolated mutable state (actors)

Purely immutable (pure functions)

Shared mutable state (threads + locks)

7979

8080

When to use parallel streams –loosely speaking

•When operations are independent, and

•Either or both

• Operations are computationally expensive

• Operations are applied to many elements of

efficiently splittable data structures

•Always measure before and after parallelizing!