Les Streams sont parmi nous

@JosePaumard

sont parmi nous

@JosePaumard

sont parmi nouspour le meilleur !

#J8Stream @JosePaumard

Pourquoi s’intéresser aux API Streams ?

Traitement de donnéesTraitement de données

• De 1998 à 2014… seul outil : l’API Collection Common collections

• À partir de 2014… un foisonnement !

• Pourquoi autant d’offres ?

• Parce que le traitement de données devient central…

• Parce que le traitement de données devient central… et complexe !

• Parce que le traitement de données devient central… et complexe ! Volumes de plus en plus importants

• Parce que le traitement de données devient central… et complexe ! Volumes de plus en plus importants Temps de réponse maîtrisés

• Parce que le traitement de données devient central… et complexe ! Volumes de plus en plus importants Temps de réponse maîtrisés Algorithmes complexes

• Le traitement de données a besoin de « primitives » de haut niveau

• Le traitement de données a besoin de « primitives » de haut niveau Qui permettent d’accéder aux données où qu’elles

soient

soient Qui exposent des fonctions (map / filter / reduce)

soient Qui exposent des fonctions (map / filter / reduce) Qui soient efficaces !

soient Qui exposent des fonctions (map / filter / reduce) Qui soient efficaces ! Éventuellement parallèles

Objet de la présentationObjet de la présentation

• Présenter trois API Exposer les concepts fondamentaux Présenter les fonctions implémentées Montrer des patterns de code (Java 8, lambdas)

• Comparer ces API Point de vue de l’utilisateur Performances !

• Proposer une grille de lecture et d’analyse Comparer des outils qui proposent des solutions

différentes pour un même problème

• Du point de vue du développeur

• Du point de vue du développeur Performance Patterns écriture / lecture

Questions ?

#J8Stream

Questions ?

#J8Stream

Les forces en présenceLes forces en présence

• Les 3 API sont : L’API Stream de Java 8

• Les 3 API sont : L’API Stream de Java 8 GS Collections

• Les 3 API sont : L’API Stream de Java 8 GS Collections RxJava

• Comparaison des performances sur un exemple commun

API Stream

API StreamAPI Stream

• Qu’est-ce qu’un Stream Java 8 ? Un objet que l’on connecte à une source

• Qu’est-ce qu’un Stream Java 8 ? Un objet que l’on connecte à une source Un objet qui ne porte pas les données qu’il traite

• Qu’est-ce qu’un Stream Java 8 ? Un objet que l’on connecte à une source Un objet qui ne porte pas les données qu’il traite Un objet qui expose le pattern map / filter / reduce

• Nouveau concept introduit dans le JDK

• ExempleList<String> liste = Arrays.asList("un", "deux", "trois") ;

liste.stream() .map(s ‐> s.toUpperCase()).max(Comparator.comparing(s ‐> s.length())).ifPresent(s ‐> System.out.println(s)) ;

liste.stream() // ouverture d’un stream de String.map(s ‐> s.toUpperCase()) .max(Comparator.comparing(s ‐> s.length())).ifPresent(s ‐> System.out.println(s)) ;

liste.stream().map(s ‐> s.toUpperCase()) // mise en majuscules des éléments.max(Comparator.comparing(s ‐> s.length())).ifPresent(s ‐> System.out.println(s)) ;

liste.stream().map(s ‐> s.toUpperCase()).max(Comparator.comparing(s ‐> s.length())) // plus longue.ifPresent(s ‐> System.out.println(s)) ;

liste.stream().map(s ‐> s.toUpperCase()).max(Comparator.comparing(s ‐> s.length())).ifPresent(s ‐> System.out.println(s)) ; // Optional !

liste.stream().map(String::toUpperCase).max(Comparator.comparing(String::length).ifPresent(System.out::println) ; // Optional !

CollectorsCollectors

• On peut collecter des donnéesList<Person> liste = ... ;

liste.stream().filter(person ‐> person.getAge() > 30).collect(Collectors.groupingBy(

Person::getAge, // key extractorCollectors.counting() // downstream collector

liste.stream().filter(person ‐> person.getAge() > 30).collect(Collectors.groupingBy(

Map<Integer, Long> map = liste.stream()

.filter(person ‐> person.getAge() > 30)

.collect(Collectors.groupingBy(

• Et on peut paralléliser les traitementsList<Person> liste = ... ;

Map<Integer, Long> map = liste.stream().parallel()

.filter(person ‐> person.getAge() > 30)

.collect(Collectors.groupingBy(

Particularisation d’un StreamParticularisation d’un Stream

• On peut le connecter à sa propre source de données Implémenter un Spliterator (~ Iterator)

Particularisation d’un StreamParticularisation d’un Stream

• On peut le connecter à sa propre source de données Implémenter un Spliterator (~ Iterator)

• On peut implémenter son propre Collector

GS Collections

IntroductionIntroduction

• API Open source, Github• Développé par Goldman Sachs• Contribution importante

https://github.com/goldmansachs/gs-collections

GS CollectionsGS Collections

• Une alternative à l’API Collection Parfois très ressemblante…

• Des interfaces supplémentaires Beaucoup ! Des méthodes supplémentaires sur les interfaces

existantes

• Nouveaux concepts ?

• Oui et non…

• Oui et non… Bag

• Oui et non… Bag MultiMap

• Oui et non… Bag MultiMap MutableSet

• Plutôt des extensions des concepts de l’API Collection

• Plutôt des extensions des concepts de l’API Collection Certains sont dans Common collections ou Guava

Structures fondamentalesStructures fondamentales

• Bag = Collection• Existe en plusieurs couleurs !

• Primitifs : IntBag, LongBag, …

• Primitifs : IntBag, LongBag, …• ImmutableBag, MutableBag

• Primitifs : IntBag, LongBag, …• ImmutableBag, MutableBag• Sorted, Unsorted

• Primitifs : IntBag, LongBag, …• ImmutableBag, MutableBag• Sorted, Unsorted• Synchronized

• Bag, List, Set, Stack, Map, Multimap

• Bag, List, Set, Stack, Map, Multimap• Nombreuses implémentations

BagBag

• Méthodes originales type CollectionBag<Person> bag = ... ;

int numberOfPaul = bag.occurenceOf(paul) ;

BagBag

int numberOfPaul = bag.occurenceOf(paul) ;int numberOfDistinct = bag.sizeDistinct() ;

BagBag

int numberOfPaul = bag.occurenceOf(paul) ;int numberOfDistinct = bag.sizeDistinct() ;

Bag<Customer> customers = bag.instanceOf(Customer.class) ;

BagBag

Partition<Person> partition = bag.partition(person ‐> person.getAge() > 30) ;

BagBag

Bag<Person> selected = partition.getSelected() ;Bag<Person> rejected = partition.getRejected() ;

BagBag

Bag<Person> selected = partition.getSelected() ;Bag<Person> rejected = partition.getRejected() ;

Bag<Person> selected = bag.select(person ‐> person.getAge() > 30) ;Bag<Person> rejected = bag.reject(person ‐> person.getAge() > 30) ;

BagBag

• Méthode forEach()Bag<Person> bag = ... ;

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

BagBag

bag.forEach(System.out::println) ; // compiler error

BagBag

public interface Iterable<T> { // java.util

default void forEach(Consumer<? super T> action) {...}}

BagBag

public interface InternalIterable<T> { // GS Collection

void forEach(Procedure<? super T> procedure) ;}

BagBag

bag.forEach((Procedure<String>)System.out::println) ;

BagBag

• Méthodes map & filterBag<Person> bag = ... ;

Bag<String> names = bag.collect(Person::getAge) ; // eq. map()Bag<String> names = bag.select(p ‐> p.getAge() > 30) ; // eq filter()

BagBag

• Méthodes map & filterBag<Person> bag = ... ;

Bag<String> names = bag.collect(Person::getAge) ; // eq. map()Bag<String> names = bag.select(p ‐> p.getAge() > 30) ; // eq filter()

Bag<T> tap(Procedure<? super T> procedure) ;

BagBag

• Méthodes map & flatMapBag<T> collect( // eq. map()

Function<? super T, ? extends V> function) {...}

Bag<T> flatCollect( // eq. flatMap()Function<? super T, ? extends Iterable<V>> function) {...}

BagBag

• Méthodes map & filter fusionnéesBag<Person> bag = ... ;

names = bag.collectIf(person ‐> person.getAge() > 30, Peson::getName) ;

BagBag

• Méthodes de réductionBag<Person> bag = ... ;

boolean b1 = bag.allSatisfy(person ‐> person.getAge() > 30) ;

boolean b2 = bag.anySatisfy(person ‐> person.getAge() > 100) ;

boolean b3 = bag.noneSatisfy(person ‐> person.getAge() < 18) ;

BagBag

Person first = bag.detect(person ‐> person.getAge() > 30) ;

Person p = bag.detectIfNone(person ‐> person.getAge() > 100, () ‐> Person.NOBODY) ;

BagBag

Person first = bag.detectWith((person, param) ‐> person.getAge() > param, 30) ;

Person p = bag.detectWithIfNone((person, param) ‐> person.getAge() > param, 100, () ‐> Person.NOBODY) ;

BagBag

• Méthodes de foldingBag<Person> bag = ... ;

long sum = bag.injectInto(0L, (sum, person) ‐> sum + person.getAge()) ;

long count = bag.injectInto(0L, (sum, person) ‐> sum + 1L) ;

BagBag

long sum = bag.injectInto(0L, (l, person) ‐> l + person.getAge()) ;

long sumOfAge = bag.sumOfInt(Person::getAge)) ;

BagBag

String names = bag.collect(Person::getName).makeString() ;

Charles, Michelle, Paul, Barbara

BagBag

• Réduction dans une chaîne de caractèresBag<Person> bag = ... ;

String names = bag.collect(Person::getAge).makeString() ;

String names = bag.collect(Person::getAge).makeString(" ; ") ;

Charles ; Michelle ; Paul ; Barbara

BagBag

String names = bag.collect(Person::getAge).makeString() ;

String names = bag.collect(Person::getAge).makeString(" ; ") ;

String names = bag.collect(Person::getAge).makeString("{", " ; ", "}") ;

{Charles ; Michelle ; Paul ; Barbara}

BagBag

bag.collect(Person::getAge).appendString(StringBuilder::new) ;

BagBag

• Réduction dans une table de hachageBag<Person> bag = ... ;

MapIterable<Integer, Bag<Person>> map = bag.aggregateBy(

Person::getAge, // key extractorFastList::new, // zero factory(list, person) ‐> { //

list.add(person) ; // merge operationreturn list ; //

BagBag

• Réduction dans une table de hachageBag<Person> bag = ... ;

MultiMap<Integer, Person> map = bag.groupBy(Person::getAge) ;

MutableMapMutableMap

• Méthodes de MapV getIfAbsentPut(K key,

Function0<? extends V> function) ;

V getIfAbsentPutWithKey(K key, Function<? super K, ? extends V> function) ;

V getIfAbsentPutWith(K key, Function<? super P, ? extends V> functionP parameter) ;

• Méthodes forEach()void forEachKeyValue(Procedure2<? super K, ? super V> procedure) ;

void forEachKey(Procedure<? super T> procedure) ;

void forEachValue(Procedure<? super T> procedure) ;

• Méthodes map

• Une table de hachage est aussi une liste de valeurs

MapIterable<K2, V2> collect(Function2<? super K, ? super V, Pair<K2, V2>> function) ;

public interface MapIterable<K, V> extends RichIterable<V> {}

PairPair

• Modélisation d’un tuplepublic interface Pair<T1, T2>

extends Serializable, Comparable<Pair<T1, T2>> {

T1 getOne();T2 getTwo();

void put(Map<T1, T2> map);

Map.Entry<T1, T2> toEntry(); // pont avec API Collection Pair<T2, T1> swap();

• Les méthodes map, filter et méthodes de réduction sont définies sur RichIterable Donc disponibles sur les tables de hachage Elles opèrent sur les valeurs

• Méthodes groupBy, retournent de nouvelles tables

reject(Predicate2<? super K, ? super V> predicate) ;

select(Predicate2<? super K, ? super V> predicate) ;

ImmutableMapImmutableMap

• Méthodes pour « ajouter » des clés / valeursMutableMapIterable<K, V> withKeyValue(K key, V value) ; // put

MutableMapIterable<K, V> withoutKey(K key) ; // remove

MultiMapMultiMap

• Concept qui ne fait pas partie de l’API Collection• Map qui associe plusieurs valeurs à une même clé Gère une collection de valeurs en interne

MultiMap<Integer, Person> map ; // GS Collections

Map<Integer, List<Person>> map ; // API Collection

MultiMapMultiMap

• La plupart des méthodes sont les mêmes• Méthodes get différentesRichIterable<V> values = multiMap.get(K key) ;

MutableMap<K, RichIterable<V>> map = multiMap.toMap() ;

Structures concurrentesStructures concurrentes

• ConcurrentHashMap Construite sur un tableau de Entry

(AtomicReferenceArray) Accès via des AtomicReference Même structure que java.util.HashMap

Opérations de type ZipOpérations de type Zip

• Permet d’appairer deux ensemblesBag<Person> bag1 = ImmutableBagImpl.of(anna, charles) ;Bag<Person> bag2 = ImmutableBagImpl.of(paul, barbara) ;

Bag<Pair<Person, Person>> friends = bag1.zip(bag2) ;

• Permet d’appairer deux ensemblesBag<Person> bag1 = ImmutableBagImpl.of(anna, charles) ;Bag<Person> bag2 = ImmutableBagImpl.of(paul, barbara) ;Bag<City> bag3 = ImmutableBagImpl.of(paris, sanFrancisco) ;

Bag<Pair<Person, City>> living = bag1.zip(bag3) ;

• Permet d’appairer deux ensemblesBag<Person> bag1 = ImmutableBagImpl.of(anna, charles) ;Bag<Person> bag2 = ImmutableBagImpl.of(paul, barbara) ;Bag<City> bag3 = ImmutableBagImpl.of(paris, sanFrancisco) ;

Bag<Pair<Person, City>> living = bag1.zip(bag3) ;

Bag<Pair<City, Integer>> cities = bag3.zipWithIndex() ;

Classes factoryClasses factory

• Factory pour Bag, List, Set, Stack, MapBag<Long> bagOfInteger = MutableBagFactoryImpl.of(1L, 2L, 5L) ;

Bag<Person> bagOfPerson = MutableBagFactoryImpl.of(persons) ;

Implémentations Implémentations

• Pas de miracle à attendre…// class FastListpublic boolean add(T newItem) {

if (this.items.length == this.size) {this.ensureCapacityForAdd();

}this.items[this.size++] = newItem;return true;

• Pas de miracle à attendre…// class FastListpublic boolean add(T newItem) {

if (this.items.length == this.size) {this.ensureCapacityForAdd();

}this.items[this.size++] = newItem;return true;

// class ArrayListpublic boolean add(E e) {

ensureCapacityInternal(size + 1);elementData[size++] = e;return true;

• Une implémentation est différente : celle des tables de hachage Fonctionne avec un tableau unique

Bizarreries Bizarreries

• Quelques bizarreriesMutableMap<K, V> map1 = map.asUnmodifiable() ; // UnmodifiableMap ?

Bilan sur l’APIBilan sur l’API

• Une API complète (complexe)• Concept de MultiMap• Des méthodes que l’on trouve sur Stream Ajoutées aux collections

• Des méthodes supplémentaires Que l’on aura peut-être sur Stream (zip)

RxJava

RxJavaRxJava

• API Open source, Github• Développé par Netflix• Version Java de ReactiveX .NET Python, Kotlin, JavaScript, Scala, Ruby, Groovy, Rust Android

https://github.com/ReactiveX/RxJava

Vue d’ensembleVue d’ensemble

• Approche différente de GS Collections• Il ne s’agit pas d’une implémentation alternative de

l’API Collection

• Mais plutôt de l’implémentation du pattern Reactor

Pattern ReactorPattern Reactor

• Set (fixed size, mutable, immutable, primitifs)• SortedSet (fixed size, mutable, immutable, primitifs)• Pool• Stack (mutable, immutable, primitifs)

http://www.dre.vanderbilt.edu/~schmidt/PDF/reactor-siemens.pdf

Notions fondamentalesNotions fondamentales

• Observable : source de données

• Observable : source de données Objet complexe ~100 méthodes statiques + ~150

méthodes non statiques

méthodes non statiques• Observer : permet d’observer un observable Objet simple : 3 méthodes

• Subscription : lien qui existe entre un observable et un observer

• Observerpublic interface Observer<T> {

public void onNext(T t);

public void onCompleted();

public void onError(Throwable e);}

• Associer un observer à un observableObservable<T> observable = ... ;

Subscription subscription = observable.subscribe(observer) ;

public interface Subscription {

public void unsubscribe();

public void isUnsubscribe();}

Création d’un ObservableCréation d’un Observable

• Constructeur non vide protégé

• Constructeur non vide protégé• On peut construire un Observable par extension en utilisant une des méthodes statiques

• Prend un producer en paramètre

• Création à partir de collectionsObservable<String> obs1 = Observable.just("one", "two", "three") ;

List<String> strings = Arrays.asList("one", "two", "three") ;Observable<String> obs2 = Observable.from(strings) ;

• Création à partir de collectionsObservable<String> obs1 = Observable.just("one", "two", "three") ;

List<String> strings = Arrays.asList("one", "two", "three") ;Observable<String> obs2 = Observable.from(strings) ;

Observable<String> empty = Observable.empty() ;Observable<String> never = Observable.never() ;Observable<String> error = Observable.<String>error(exception) ;

• Création de sériesObservable<Long> longs = Observable.range(1L, 100L) ;

// intervalObservable<Long> timeSerie1 =

Observable.interval(1L, TimeUnit.MILLISECONDS) ; // a serie of longs

// initial delay, then intervalObservable<Long> timeSerie2 =

Observable.timer(10L, 1L, TimeUnit.MILLISECONDS) ; // one 0

• Méthode using()public final static <T, Resource> Observable<T> using(

final Func0<Resource> resourceFactory, // producerfinal Func1<Resource, Observable<T>> observableFactory, // functionfinal Action1<? super Resource> disposeAction // consumer

Notion de SchedulerNotion de Scheduler

• Ces méthodes peuvent prendre un autre paramètre

• Scheduler : interface• Associée à la factory Schedulers

Observable<Long> longs = Observable.range(0L, 100L, scheduler) ;

• Factory Schedulerspublic final class Schedulers {

public static Scheduler immediate() {...} // immediate, same thread

public static Scheduler newThread() {...} // new thread

public static Scheduler computation() {...} // computation ES

public static Scheduler io() {...} // IO growing ES}

• Factory Schedulerspublic final class Schedulers {

public static Scheduler trampoline() {...} // queued in the current// thread

public static Scheduler test() {...}

public static Scheduler fromExecutor(Executor executor) {...}}

• RxJava permet de créer des Observable dans différents threads

• Certains pools sont spécialisés : IO, Computation• On peut passer ses propres pools

• Les observateurs sont appelés dans les threads des observables

• La classe Scheduler n’est pas immédiate à étendre

• On utilise from() pour les threads IHM (Swing, JavaFX)

@JosePaumard

Pause café ?

#J8Stream

@JosePaumard

Pause café !

#J8Stream

Encore du RxJavaPatterns

Extensions de StreamPerformance

@JosePaumard

sont parmi nous

Observable : méthodes statiques Observable : méthodes statiques

• Série de méthodes statiques de combinaisons d’Observable en un seul

Observable de listesObservable de listes

• amb() : prend une liste d’Observable et suit le premier qui parle

©RxJava

• combineLatest() : applique une fonction aux deux derniers éléments émis

©RxJava

F(O1, O2)

• zip() : équivalent d’un combine, mais prend les éléments un par un

©RxJava

F(O1, O2)

• concat() : émet O1 puis O2, sans les mélanger• merge() : émet O1 et O2, s’arrête sur erreur

©RxJava©RxJava

• concat() : émet O1 puis O2, sans les mélanger• merge() : émet O1 et O2, s’arrête avec le premier• mergeDelayError() : reporte l’erreur à la fin

©RxJava

• sequenceEqual() : compare deux séquences

©RxJava

boolean

Observable d’ObservableObservable d’Observable

• switchOnNext() : un peu spécial…

• Paramètres des méthodes précédentes :public final static <T> Observable<T> merge(

Iterable<Observable<T>> listOfSequences) { }

public final static <T> Observable<T> merge(Observable<Observable<T>> sequenceOfSequences) { }

©RxJava

Observable<Observable<T>> sequenceOfSequences) { }

©RxJava

• switchOnNext() : prend le nouvel Observable

©RxJava

public final static <T> Observable<T> switchOnNext(Observable<Observable<T>> sequenceOfSequences) { }

ExemplesExemples

• Un petit cas simpleObservable<Integer> range1To100 = Observable.range(1L, 100L) ;manyStrings.subscribe(System.out::println) ;

ExemplesExemples

• Un petit cas simpleObservable<Integer> range1To100 = Observable.range(1L, 100L) ;manyStrings.subscribe(System.out::println) ;

> 1 2 3 4 ... 100

ExemplesExemples

• Un petit cas un peu plus durObservable<Integer> timer = Observable.timer(1, TimeUnit.SECONDS) ;timer.subscribe(System.out::println) ;

ExemplesExemples

• Un petit cas un peu plus dur

• N’affiche rien…

Observable<Integer> timer = Observable.timer(1, TimeUnit.SECONDS) ;timer.subscribe(System.out::println) ;

ExemplesExemples

• Un petit cas un peu plus durObservable<Integer> timer = Observable.timer(1, TimeUnit.SECONDS) ;timer.subscribe(() ‐> {

System.out.println(Thread.currentThread().getName() + " " + Thread.currentThread().isDaemon()) ;

}) ;Thread.sleep(2) ;

ExemplesExemples

• Un petit cas un peu plus durObservable<Integer> timer = Observable.timer(1, TimeUnit.SECONDS) ;timer.subscribe(() ‐> {

System.out.println(Thread.currentThread().getName() + " " + Thread.currentThread().isDaemon()) ;

}) ;Thread.sleep(2) ;

> RxComputationThreadPool‐1 ‐ true

ExemplesExemples

• Encore un peu plus durObservable<Integer> range1To100 = Observable.range(1, 100) ;

Observable<String> manyStrings = Observable.combineLatest(

range1To100, Observable.just("one"),(integer, string) ‐> string) ;

ExemplesExemples

• Encore un peu plus durObservable<Integer> range1To100 = Observable.range(1, 100) ;

range1To100, Observable.just("one"),(integer, string) ‐> string) ;

> one (et c’est tout)

Combines two source Observables by emitting an item that aggregatesthe latest values of each of the source Observables each time an item is received from either of the source Observables, where this aggregation is defined by a specified function.

ExemplesExemples

• Encore un peu plus durObservable<Integer> serie = Observable.interval(3, TimeUnit.MILLISECONDS) ;

serie, Observable.just("one"),(integer, string) ‐> string) ;

ExemplesExemples

• Encore un peu plus durObservable<Integer> serie = Observable.interval(3, TimeUnit.MILLISECONDS) ;

serie, Observable.just("one"),(integer, string) ‐> string) ;

> one one one one one one ...

Observables hot & coldObservables hot & cold

• Deux mécanismes d’Observable : Ceux qui émettent des données lorsqu’elles sont

consommées = cold observables Ceux qui émettent des données, qu’elles soient

consommées ou pas = hot observables• Un hot observable peut générer ses données

indépendamment de ses observateurs

Méthodes d’instance

Opérations callbackOpérations callback

• Retournent un Observable<T>

• ~10 versions doOn*

doOnNext(Action1<T> onNext) { } // consumer

Opérations booléennesOpérations booléennes

• Retournent un Observable<Boolean>

• Retournent un Observable<T> vide ou singleton

exists(Func1<T, Boolean> predicate) { }elementAt(int index) { }

ignoreElement() { } // returns an empty Observablesingle() { } // completes on error or transmitssingle(Func1<T, Boolean> predicate) { } // returns the item or error

Opérations de mappingOpérations de mapping

• Retournent un Observable<R>map(Func1<T, R> mapping) { } cast(Class<R> clazz) { } // casts the elementstimestamp() { } // wraps the elements in a timestamped object

zipWith(Observable other, Func2<T, U, R> zipFunction) { } // BiFunction

flatMap(Func1<T, Observable<R>> mapping) { }

Opérations de filtrageOpérations de filtrage

• Retourne un Observable<T>filter(Func1<T, Boolean> predicate) { }

Opérations de sélectionOpérations de sélection

• Retournent un Observable<R>sample(long period, TimeUnit timeUnit) { }

©RxJava

sampler

Opérations de sélectionOpérations de sélection

• Retournent un Observable<R>sample(Observable sampler) { } // samples on emission & completion

©RxJava

sampler

Remarque sur le tempsRemarque sur le temps

• RxJava peut mesurer le temps « en réel »• Peut aussi utiliser un Observable comme mesure du

temps Appels à onNext() et onComplete()

Sélection d’élémentsSélection d’éléments

• Retournent Observable<T>

• Exception en cas de timeout entre deux émissions

first() { }last() { }skip(int n) { }limit(int n) { }take(int n) { }

timeout(long n, TimeUnit timeUnit) { }timeout(Func0<Observable> firstTimeoutSelector, // producer

Func1<T, Observable<V>> timeoutSelector) { } // function

Méthodes de réductionMéthodes de réduction

• Retourne un Observable<> singleton

• Retourne un Observable<T>

all(Func1<T, Boolean> predicate) { } // Observable<Boolean>count() { } // Observable<Long>reduce(Func2<T, T, T> accumulator) { } // Observable<T>

scan(Func2<T, T, T> accumulator) { } // Observable<T>

forEach(Action1<T> consumer) { } // void

Méthodes de collectionMéthodes de collection

• Méthode collectcollect(Func0<R> stateFactory, // producer

Action2<R, T> collector) { } // BiConsumer

• Méthode collect

• Construit une Map<T, List<String>>

collect(Func0<R> stateFactory, // producerAction2<R, T> collector) { } // BiConsumer

collect(() ‐> new ArrayList<String>(), // producerArrayList::add) { } // BiConsumer

• Retournent des Observable<> singletontoList() { } // Observable<List<T>>toSortedList() { } // Observable<List<T>>

toMultimap(Func1<T, K> keySelector, // Observable<Func1<T, V> valueSelector) { } // Map<K, Collection<T>>

toMap(Func1<T, K> keySelector) { } // Observable<Map<K, T>>

Opérations join / groupJoinOpérations join / groupJoin

• Retourne un Observable<GroupedObservable<K, T>>

• GroupedObservable : observable avec une clé

groupBy(Func1<T, K> keySelector) { } // function

• Combine deux observables, sur un timerpublic final <T2, D1, D2, R> Observable<R> groupJoin(

Observable<T2> right, Func1<T, Observable<D1>> leftDuration, // functionFunc1<T2, Observable<D2>> rightDuration, // functionFunc2<T, Observable<T2>, R> resultSelector // bifunction

• Les fenêtres de temps sont définies par des observables Elles démarrent lorsque l’observable démarre Et se ferment lorsque l’observable émet un objet ou

s’arrête

• Combine deux observables

Méthodes de debugMéthodes de debug

• Deux méthodes, font du mapping

• Notification : objet qui englobe des meta-données en plus

materialize() { } // Observable<Notification<T>>

Méthodes de debugMéthodes de debug

• Deux méthodes, font du mapping

• Notification : objet qui englobe des métadonnées en plus des objets de l’Observable

materialize() { } // Observable<Notification<T>>

dematerialize() { } // Observable<T>

Méthodes synchronesMéthodes synchrones

• Associent des opérations avec une horlogedelay(long delay, TimeUnit timeUnit) ; // Observable<T>

• Associent des opérations avec une horloge

• Ou avec un autre Observable

delay(long delay, TimeUnit timeUnit) ; // Observable<T>

delay(Func1<T, Observable func1) ; // Observable<T>

• Limite le nombre d’éléments émisdebounce(long delay, TimeUnit timeUnit) ; // Observable<T>

©RxJava

• Limite le nombre d’éléments émis

• Peut aussi se caler sur les événements d’un Observable

debounce(long delay, TimeUnit timeUnit) ; // Observable<T>

debounce(Func1<T, Observable func1) ; // Observable<T>

• Mesure du temps entre événements

• TimeInterval : wrapper pour la durée en ms et la valeur émise

timeInterval() { } // Observable<TimeInterval<T>>

• Méthodes throttle()throttleFirst(long windowDuration, TimeUnit unit) { } // Observable<T>

throttleLast(long windowDuration, TimeUnit unit) { }

throttleWithTimeout(long windowDuration, TimeUnit unit) { }

Backpressure

Méthodes backpressureMéthodes backpressure

• Problème : une source peut générer « trop » d’éléments Par rapport à la vitesse des consommateurs

• Certaines méthodes permettent de « sauter » des éléments

• Problème : une source peut émettre « trop » d’éléments Par rapport à la vitesse des consommateurs

• Ne peut pas arriver avec les cold observables

• Problème : une source peut émettre « trop » d’éléments Par rapport à la vitesse des consommateurs

• Ne peut pas arriver avec les cold observables

• Un observable cold peut devenir hot…

• Backpressure : consiste à ralentir le rythme d’émission

• Méthodes utiles : sample() et debounce()• Méthode buffer() : stocke les éléments avant de les

émettre

• Méthodes buffer()buffer(int size) { } // Observable<List<T>>

buffer(long timeSpan, TimeUnit unit) { } // Observable<List<T>>buffer(long timeSpan, TimeUnit unit, int maxSize) { }

buffer(Observable<O> bufferOpenings, // Openings eventsFunc1<O, Observable<C>> bufferClosings) { } // Closings events

• Méthodes window()window(int size) { } // Observable<Observable<T>>

window(long timeSpan, TimeUnit unit) { } // Observable<Observable<T>>window(long timeSpan, TimeUnit unit, int maxSize) { }

window(Observable<O> bufferOpenings, // Openings eventsFunc1<O, Observable<C>> bufferClosings) { } // Closings events

Méthodes repeat & retryMéthodes repeat & retry

• repeat() : répète l’émission d’objets indéfinimentrepeat() { } // Observable<T>repeat(long times) { } // Observable<T>

• repeat() : répète l’émission d’objets indéfiniment

• Quand l’observable appelle onComplete(), le notification handler est invoqué

repeat() { } // Observable<T>repeat(long times) { } // Observable<T>

repeatWhen(Func1<Observable<Void>>, Observable<?>> notificationHandler

• repeat() : répète l’émission d’objets indéfiniment

• Sur cet Observable, le notification handler peut alors invoquer : onComplete() ou onError(), ce qui déclenche le même

appel sur l’Observable source onNext(), ce qui déclenche la répétition

• retry() : répète l’émission d’objets sur erreurretry() { } // Observable<T>retry(long times) { } // Observable<T>

• retry() : répète l’émission d’objets sur erreur

• Quand l’observable appelle onError(), le notification handler est invoqué avec l’exception

retry() { } // Observable<T>retry(long times) { } // Observable<T>

retryWhen(Func1<Observable<Throwable>>, Observable<?>> notificationHandler

• Sur cet Observable, le notification handler peut alors invoquer : onComplete() ou onError(), ce qui déclenche le même

appel sur l’Observable source onNext(), ce qui déclenche la répétition

Plusieurs ObserverPlusieurs Observer

• À la différence des Stream de Java 8, un Observable peut avoir plusieurs observateurs

• Ce qui peut poser des problèmes…

• Cas d’un cold Observable : La consommation par plusieurs observateurs ne pose

pas de problème• Cas d’un hot Observable : Un deuxième observateur peut « manquer » les

premières valeurs

• Méthode cache()

• Permet de ne pas « manquer » les premières valeurs

cache() { }cache(int capacity) { }

• Notion de ConnectableObserverObservable<T> observable = ... ;ConnectableObservable<T> connectable = observable.publish() ;

• Notion de ConnectableObserver

• L’enregistrement d’observateurs ne déclenche pas la production des éléments

Observable<T> observable = ... ;ConnectableObservable<T> connectable = observable.publish() ;

• Notion de ConnectableObserver

• L’enregistrement d’observateurs ne déclenche pas la production des éléments

• Pour cela : appeler connect()

Observable<T> observable = ... ;ConnectableObservable<T> connectable = observable.publish() ;

connectable.publish() ;

Bilan sur RxJavaBilan sur RxJava

• API Complète (complexe)• Permet de gérer le lancement de traitements dans

différents pools de threads

Bilan sur RxJavaBilan sur RxJava

• API Complète (complexe)• Permet de gérer le lancement de traitements dans

différents pools de threads• Permet de synchroniser les opérations Sur une horloge Sur des références applicatives

Ponts entre Java 8 Stream /

GS Collections / RxJava

Le meilleur des deux mondes ?Le meilleur des deux mondes ?

• Connecter GS Collections et Java 8 Stream ?= connecter Stream et Iterable

Le meilleur des deux mondes ?Le meilleur des deux mondes ?

• Connecter GS Collections et Java 8 Stream ?= connecter Stream et Iterable

• Connecter Java 8 Stream et RxJava ? = connecter Stream et Observable

Spliterator et IteratorSpliterator et Iterator

• Les collections sont construites sur des Iterator• Les Stream sont construits sur des Spliterator

Iterator<T> iterator = ... ;Spliterator<T> spliterator =

Spliterators.spliteratorUnknownSize(iterator, 0) ;

Spliterator<T> spliterator = Spliterators.spliterator(iterator, size, 0) ;

Spliterator et IteratorSpliterator et Iterator

• Les collections sont construites sur des Iterator• Les Stream sont construits sur des Spliterator

Spliterator<T> spliterator = ... ;

Iterator<T> iterator = Spliterators.iterator(spliterator) ;

Iterator<T> iterator = ... ;

Iterable<T> iterable = () ‐> iterator ;

GS Collections & StreamGS Collections & Stream

• Le problème consiste à passer d’une Collection à un Stream (et réciproquement)

GS Collections & StreamGS Collections & Stream

• Le problème consiste à passer d’une Collection à un Stream (et réciproquement)

• Donc d’un Iterator à un Spliterator (et réciproquement)

RxJava & StreamRxJava & Stream

• Si l’on a un Iterator : facile !Iterator<T> iterator = ... ;

Observable<T> observable = Observable.from(() ‐> iterator) ;

• Si l’on a un Iterator : facile !

• Si l’on a un Spliterator : facile !

Iterator<T> iterator = ... ;

Observable<T> observable = Observable.from(() ‐> iterator) ;

Spliterator<T> spliterator = ... ;

Observable<T> observable = Observable.from(() ‐> Spliterators.iterator(spliterator)) ;

• Donc si l’on a un Stream, on peut facilement construire un Observable

• L’inverse ?

• L’inverse ? On peut construire un Iterator sur un Observable Puis un Spliterator sur un Iterator

• Implémenter un Iterator : Deux méthodes next() et hasNext() La méthode remove() et une méthode par défaut

• Un Iterator « tire » les données d’une source• Alors qu’un Observable « pousse » les données vers

des callbacks

• Un Iterator « tire » les données d’une source• Alors qu’un Observable « pousse » les données vers

des callbacks

• Il nous faut donc une adaptation entre les deux

• On peut s’inspirer du JDKpublic static<T> Iterator<T> iterator(Spliterator<? extends T> spliterator) {

class Adapter implements Iterator<T>, Consumer<T> {// implementation

return new Adapter() ;}

• On peut s’inspirer du JDK

class Adapter implements Iterator<T>, Consumer<T> {boolean valueReady = false ;T nextElement;

public void accept(T t) {valueReady = true ;nextElement = t ;

public boolean hasNext() {if (!valueReady)

spliterator.tryAdvance(this) ; // calls accept()return valueReady ;

public T next() {if (!valueReady && !hasNext())

throw new NoSuchElementException() ;else {

valueReady = false ;return nextElement ;

• On peut s’inspirer du JDKpublic static<T> Iterator<T> of(Observable<? extends T> obsevable) {

class Adapter implements Iterator<T> {// implementation

return new Adapter() ;}

• On peut s’inspirer du JDK

class Adapter implements Iterator<T>, Consumer<T> {boolean valueReady = false ;T nextElement;

public void accept(T t) {valueReady = true ;nextElement = t ;

public boolean hasNext() {return valueReady ;

public T next() {if (!valueReady && !hasNext())

throw new NoSuchElementException() ;else {

valueReady = false ;return nextElement ;

• On peut s’inspirer du JDKclass Adapter implements Iterator<T>, Consumer<T> {boolean valueReady = false ;T nextElement;

public void accept(T t) {observable.subscribe(

element ‐> nextElement = element, // onNextexception ‐> valueReady = false, // onError() ‐> valueReady = false // onComplete

// next

// hasNext}

• On peut s’inspirer du JDKclass Adapter implements Iterator<T>, Consumer<T> {AtomicBoolean valueReady = new AtomicBoolean(false) ;AtomicReference<T> nextElement = new AtomicReference() ;

element ‐> nextElement.set(element), // onNextexception ‐> valueReady.set(false), // onError() ‐> valueReady.set(false) // onComplete

// next

// hasNext}

• Peut mieux faire ? interface Wrapper<E> {

E get() ;

Wrapper<Boolean> wb = () ‐> true ;

E get() ;

Wrapper<Boolean> wb = () ‐> true ;Action1<Boolean> onNext = b ‐> wb.set(b) ; // should return Wrapper<T>

E get() ;

public default Wrapper<E> set(E e) {// should return e

E get() ;

public default Wrapper<E> set(E e) {return () ‐> e ;

• On peut s’inspirer du JDKclass Adapter implements Iterator<T>, Consumer<T> {Wrapper<Boolean> valueReady = () ‐> false ;Wrapper<T> nextElement ;

element ‐> nextElement.set(element), // onNextexception ‐> valueReady.set(false), // onError() ‐> valueReady.set(false) // onComplete

// next

// hasNext}

• On peut construire un Iterator sur un Observable Et donc un Spliterator sur un Observable

• Fonctionne sur les cold Observable

Spliterators avancésSpliterators avancés

• Peut-on construire des Stream à la façon des Observable ?

• Partant d’un Stream [1, 2, 3, 4, 5, …]

• Partant d’un Stream [1, 2, 3, 4, 5, …] [[1, 2, 3], [4, 5, 6], [7, 8, 9], …] ?

• Partant d’un Stream [1, 2, 3, 4, 5, …] [[1, 2, 3], [4, 5, 6], [7, 8, 9], …] ? [[1, 2, 3], [2, 3, 4], [3, 4, 5], …] ?

• Deux Stream [10, 11, 12, …] et [50, 51, 52, …]

• Deux Stream [10, 11, 12, …] et [50, 51, 52, …] [10, 50, 11, 51, 12, 52, …] ?

• Deux Stream [10, 11, 12, …] et [50, 51, 52, …] [10, 50, 11, 51, 12, 52, …] ? [Pair(10, 50), Pair(11, 51), Pair(12, 52), …] ?

• Trois méthodes pour un Spliterator tryAdvance(Consumer) trySplit(), retourne un sous-spliterator estimateSize()

GroupingSpliteratorGroupingSpliterator

• [1, 2, 3, 4, 5, …] -> [[1, 2, 3], [4, 5, 6], [7, 8, 9], …]public class GroupingSpliterator<E> implements Spliterator<Stream<E>> {

private final long grouping ;private final Spliterator<E> spliterator ;

// implementation}

• [1, 2, 3, 4, 5, …] -> [[1, 2, 3], [4, 5, 6], [7, 8, 9], …]public boolean tryAdvance(Consumer<? super Stream<E>> action) {

boolean finished = false ;Stream.Builder<E> builder = Stream.builder() ;for (int i = 0 ; i < grouping ; i++) {

if (!spliterator.tryAdvance(element ‐> { builder.add(element) ; })) {finished = true ;

}}Stream<E> subStream = subBuilder.build() ;action.accept(subStream) ;return !finished ;

• [1, 2, 3, 4, 5, …] -> [[1, 2, 3], [4, 5, 6], [7, 8, 9], …]public Spliterator<Stream<E>> trySplit() {

Spliterator<E> spliterator = this.spliterator.trySplit() ;return new GroupingSpliterator<E>(spliterator, grouping) ;

• [1, 2, 3, 4, 5, …] -> [[1, 2, 3], [4, 5, 6], [7, 8, 9], …]public long estimateSize() {

return spliterator.estimateSize() / grouping ;}

RollingSpliteratorRollingSpliterator

• [1, 2, 3, 4, 5, …] -> [[1, 2, 3], [2, 3, 4], [3, 4, 5], …]

• [1, 2, 3, 4, 5, …] -> [[1, 2, 3], [4, 5, 6], [7, 8, 9], …]public class RollingSpliterator<E> implements Spliterator<Stream<E>> {

private final int grouping ;private final Spliterator<E> spliterator ;private Object [] buffer ; // we cant create arrays of Eprivate AtomicInteger bufferWriteIndex = new AtomicInteger(0) ;private AtomicInteger bufferReadIndex = new AtomicInteger(0) ;

// implementation}

• [1, 2, 3, 4, 5, …] -> [[1, 2, 3], [4, 5, 6], [7, 8, 9], …]

public boolean tryAdvance(Consumer<? super Stream<E>> action) {boolean finished = false ;

if (bufferWriteIndex.get() == bufferReadIndex.get()) {for (int i = 0 ; i < grouping ; i++) {

if (!advanceSpliterator()) {finished = true ;

}if (!advanceSpliterator()) {

finished = true ;}

Stream<E> subStream = buildSubstream() ;action.accept(subStream) ;return !finished ;

• [1, 2, 3, 4, 5, …] -> [[1, 2, 3], [4, 5, 6], [7, 8, 9], …]private boolean advanceSpliterator() {

return spliterator.tryAdvance(element ‐> {

buffer[bufferWriteIndex.get() % buffer.length] = element ; bufferWriteIndex.incrementAndGet() ;

• [1, 2, 3, 4, 5, …] -> [[1, 2, 3], [4, 5, 6], [7, 8, 9], …]public Spliterator<Stream<E>> trySplit() {

return new RollingSpliterator<E>(spliterator.trySplit(), grouping) ;}

public long estimateSize() {return spliterator.estimateSize() ‐ grouping ;

ZippingSpliteratorZippingSpliterator

• [1, 2, 3, …], [a, b, c, …] -> [F(1, a), F(2, b), F(3, c), …]public class ZippingSpliterator<E1, E2, R> implements Spliterator<R> {

private final Spliterator<E1> spliterator1 ;private final Spliterator<E2> spliterator2 ;private final BiFunction<E1, E2, R> tranform ;

// implementation}

• [1, 2, 3, …], [a, b, c, …] -> [F(1, a), F(2, b), F(3, c), …]public boolean tryAdvance(Consumer<? super R> action) {

return spliterator1.tryAdvance(e1 ‐> {

spliterator2.tryAdvance(e2 ‐> {action.accept(tranform.apply(e1, e2)) ;

}) ;}) ;

• [1, 2, 3, …], [a, b, c, …] -> [F(1, a), F(2, b), F(3, c), …]public Spliterator<R> trySplit() {

return new ZippingSpliterator<E1, E2, R>(spliterator1.trySplit(), spliterator2.trySplit(), tranform) ;

public long estimateSize() {return this.spliterator1.estimateSize() ;

PairingSpliteratorPairingSpliterator

• [1, 2, 3, …], [a, b, c, …] -> [(1, a), (2, b), (3, c), …]public class PairingSpliterator<E1, E2> extends ZippingSpliterator<E1, E2, Pair<E1, E2>> {

public PairingSpliterator(Spliterator<E1> spliterator1, Spliterator<E2> spliterator2) {

super(spliterator1, spliterator2, (e1, e2) ‐> new Pair<E1, E2>(e1, e2)) ;

Stream vs RxJavaStream vs RxJava

• Pour la partie cold : on peut l’implémenter avec des Stream

• Pour la partie hot : on peut interfacer les deux API

Comparaisons

ComparaisonsComparaisons

• Cas applicatif unique• Implémentés avec les trois API

• Comparaisons des patterns• Mesure des temps de traitement avec JMH

Shakespeare joue an ScrabbleShakespeare joue an Scrabble

• « Shakespeare joue au Scrabble »• Ensemble de mots Les mots utilisés par Shakespeare Les mots autorisés au Scrabble

• Question : quel aurait été le meilleur mot que Shakespeare aurait pu jouer ?

Shakespeare joue an ScrabbleShakespeare joue an Scrabble

• Comparaison des principaux patterns• Puis comparaison des performances globales

Histogramme des lettresHistogramme des lettres

• Java Stream GS Collections Rx Java// Histogram of the letters in a given wordFunction<String, Map<Integer, Long>> histoOfLetters =

word ‐> word.chars().boxed().collect(

Collectors.groupingBy(Function.identity(),Collectors.counting()

• Java Stream GS Collections Rx Java// Histogram of the letters in a given wordFunction<String, MutableMap<Integer, Long>> histoOfLetters =

word ‐> new CharArrayList(word.toCharArray()).collect(c ‐> new Integer((int)c))// .groupBy(letter ‐> letter) ;.aggregateBy(

letter ‐> letter, () ‐> 0L, (value, letter) ‐> { return value + 1 ; }

• Java Stream GS Collections Rx Java// Histogram of the letters in a given wordFunc1<String, Observable<HashMap<Integer, LongWrapper>>> histoOfLetters =

word ‐> toIntegerObservable.call(word).collect(

() ‐> new HashMap<Integer, LongWrapper>(), (HashMap<Integer, LongWrapper> map, Integer value) ‐> {

LongWrapper newValue = map.get(value) ;if (newValue == null) {

newValue = () ‐> 0L ;}map.put(value, newValue.incAndSet()) ;

• Java Stream GS Collections Rx Javainterface LongWrapper {

long get() ;

public default LongWrapper set(long l) {return () ‐> l ;

public default LongWrapper incAndSet() {return () ‐> get() + 1L ;

public default LongWrapper add(LongWrapper other) {return () ‐> get() + other.get() ;

Nombre de blancs pour une lettreNombre de blancs pour une lettre

• Java Stream GS Collections Rx Java// number of blanks for a given letterToLongFunction<Map.Entry<Integer, Long>> blank =

entry ‐> Long.max(

0L, entry.getValue() ‐

scrabbleAvailableLetters[entry.getKey() ‐ 'a']) ;

• Java Stream GS Collections Rx Java// number of blanks for a given letterLongFunction<Map.Entry<Integer, Long>> blank =

entry ‐> Long.max(

0L, entry.getValue() ‐

scrabbleAvailableLetters[entry.getKey() ‐ 'a']) ;

• Java Stream GS Collections Rx Java// number of blanks for a given letterFunc1<Entry<Integer, LongWrapper>, Observable<Long>> blank =

entry ‐>Observable.just(

Long.max(0L, entry.getValue().get() ‐

scrabbleAvailableLetters[entry.getKey() ‐ 'a']

Nombre de blancs pour un motNombre de blancs pour un mot

• Java Stream GS Collections Rx Java// number of blanks for a given wordFunction<String, Long> nBlanks =

word ‐> histoOfLetters.apply(word).entrySet().stream().mapToLong(blank).sum();

• Java Stream GS Collections Rx Java// number of blanks for a given wordFunction<String, Long> nBlanks =

word ‐> UnifiedSet.newSet(histoOfLetters.valueOf(word)

.entrySet()).sumOfLong(blank) ;

• Java Stream GS Collections Rx Java// number of blanks for a given wordFunc1<String, Observable<Long>> nBlanks =

word ‐> histoOfLetters.call(word).flatMap(map ‐> Observable.from(() ‐> map.entrySet().iterator())).flatMap(blank).reduce(Long::sum) ;

Prédicat sur 2 blancsPrédicat sur 2 blancs

• Java Stream GS Collections Rx Java// can a word be written with 2 blanks?Predicate<String> checkBlanks = word ‐> nBlanks.apply(word) <= 2 ;

// can a word be written with 2 blanks?Predicate<String> checkBlanks = word ‐> nBlanks.valueOf(word) <= 2 ;

// can a word be written with 2 blanks?Func1<String, Observable<Boolean>> checkBlanks =

word ‐> nBlanks.call(word).flatMap(l ‐> Observable.just(l <= 2L)) ;

Bonus lettre doublée – 1Bonus lettre doublée – 1

• Java Stream GS Collections Rx Java// Placing the word on the board// Building the streams of first and last lettersFunction<String, IntStream> first3 =

word ‐> word.chars().limit(3);

• Java Stream GS Collections Rx Java// Placing the word on the board// Building the streams of first and last lettersFunction<String, MutableList<Integer>> first3 =

word ‐> new CharArrayList(word.toCharArray()).collect(c ‐> (int)c) ;.subList(0, Integer.min(list.size(), 3)) ;

• Java Stream GS Collections Rx Java// Placing the word on the board// Building the streams of first and last lettersFunc1<String, Observable<Integer>> first3 =

word ‐> Observable.from(

IterableSpliterator.of(word.chars().boxed().limit(3).spliterator()

• Java Stream GS Collections Rx Java// Bonus for double letterToIntFunction<String> bonusForDoubleLetter =

word ‐> Stream.of(first3.apply(word), last3.apply(word)).flatMapToInt(Function.identity()).map(scoreOfALetter).max().orElse(0) ;

• Java Stream GS Collections Rx Java// Bonus for double letterFunction<String, MutableList<Integer>> toBeMaxed =

word ‐> { MutableList<Integer> list = first3.valueOf(word) ;list.addAll(last3.valueOf(word)) ;return list ;

IntFunction<String> bonusForDoubleLetter =word ‐> toBeMaxed.valueOf(word)

.collect(scoreOfALetter)

.max() ;

• Java Stream GS Collections Rx Java// Bonus for double letterFunc1<String, Observable<Integer>> bonusForDoubleLetter =

word ‐> Observable.just(first3.call(word), last3.call(word)).flatMap(observable ‐> observable).flatMap(scoreOfALetter).reduce(Integer::max) ;

Score d’un motScore d’un mot

• Java Stream GS Collections Rx Java// score of the word put on the boardFunction<String, Integer> score3 =

word ‐> 2*(score2.apply(word) + bonusForDoubleLetter.applyAsInt(word))+ (word.length() == 7 ? 50 : 0);

• Java Stream GS Collections Rx Java// score of the word put on the boardFunction<String, Integer> score3 =

word ‐>2*(score2.valueOf(word) + bonusForDoubleLetter.intValueOf(word))+ (word.length() == 7 ? 50 : 0);

• Java Stream GS Collections Rx Java// score of the word put on the boardFunc1<String, Observable<Integer>> score3 =

word ‐>Observable.just(

score2.call(word), score2.call(word), bonusForDoubleLetter.call(word), bonusForDoubleLetter.call(word), Observable.just(word.length() == 7 ? 50 : 0)

).flatMap(observable ‐> observable).reduce(Integer::sum) ;

Histogramme des scoresHistogramme des scores

• Java Stream GS Collections Rx JavaFunction<Function<String, Integer>, Map<Integer, List<String>>>buildHistoOnScore =

score ‐> shakespeareWords.stream().parallel().filter(scrabbleWords::contains).filter(checkBlanks).collect(

Collectors.groupingBy(score, () ‐> new TreeMap<>(Comparator.reverseOrder()), Collectors.toList()

• Java Stream GS Collections Rx JavaFunction<

Function<String, Integer>, MutableMap<Integer, MutableList<String>>>buildHistoOnScore =

score ‐> shakespeareWords.select(scrabbleWords::contains).select(checkBlanks).aggregateBy(

score, FastList::new, (list, value) ‐> { list.add(value) ; return list ; }

• Java Stream GS Collections Rx JavaFunc1<Func1<String, Observable<Integer>>, Observable<TreeMap<Integer, List<String>>>> buildHistoOnScore =

score ‐> Observable.from(() ‐> shakespeareWords.iterator()).filter(scrabbleWords::contains).filter(word ‐> checkBlanks.call(word).toBlocking().first()).collect(

() ‐> new TreeMap<Integer, List<String>>(Comparator.reverseOrder()), (TreeMap<Integer, List<String>> map, String word) ‐> {

Integer key = score.call(word).toBlocking().first() ;List<String> list = map.get(key) ;if (list == null) {

list = new ArrayList<String>() ;map.put(key, list) ;

}list.add(word) ;

Meilleurs motsMeilleurs mots

• Java Stream GS Collections Rx Java// best key / value pairsList<Entry<Integer, List<String>>> finalList =

buildHistoOnScore.apply(score3).entrySet().stream().limit(3).collect(Collectors.toList()) ;

• Java Stream GS Collections Rx Java// best key / value pairsMutableList<Entry<Integer, MutableList<String>>> finalList =

new FastList<Map.Entry<Integer,MutableList<String>>>(new TreeSortedMap<Integer, MutableList<String>>(

Comparator.reverseOrder(), buildHistoOnScore.valueOf(score3)

).entrySet()

).subList(0, 3) ;

• Java Stream GS Collections Rx Java// best key / value pairsList<Entry<Integer, List<String>>> finalList2 =

buildHistoOnScore.call(score3).flatMap(map ‐> Observable.from(() ‐> map.entrySet().iterator())).take(3).collect(

() ‐> new ArrayList<Entry<Integer, List<String>>>(), (list, entry) ‐> { list.add(entry) ; }

).toBlocking().first() ;

• Java Stream GS Collections Rx Java// best key / value pairsCountDownLatch latch = new CountDownLatch(3) ;

buildHistoOnScore.call(score3).flatMap(map ‐> Observable.from(() ‐> map.entrySet().iterator())).take(3).collect(

() ‐> new ArrayList<Entry<Integer, List<String>>>(), (list, entry) ‐> { list.add(entry) ; latch.countDown() ; }

).forEach(...) ;

latch.await() ;

1er bilan : patterns1er bilan : patterns

• Java 8 Stream donne les patterns les plus simples

• Java 8 Stream donne les patterns les plus simples• Java 8 Stream & GS Collections se ressemblent

• Java 8 Stream donne les patterns les plus simples• Java 8 Stream & GS Collections se ressemblent• RxJava fait le choix du « tout flatMap » ce qui mène à

des patterns inutilement lourds

Performances Performances

• Utilisation de JMH• Outil standard de mesure de performances du JDK• Développé dans le cadre de l’Open JDK• Aleksey Shipilev http://shipilev.net/• https://twitter.com/shipilev

http://openjdk.java.net/projects/code-tools/jcstress/

http://openjdk.java.net/projects/code-tools/jmh/

https://www.parleys.com/tutorial/java-microbenchmark-harness-the-lesser-two-evils

• Utilisation de JMH• Outil standard de mesure de performances du JDK• Développé dans le cadre de l’Open JDK• Aleksey Shipilev http://shipilev.net/• https://twitter.com/shipilev

http://openjdk.java.net/projects/code-tools/jcstress/

http://openjdk.java.net/projects/code-tools/jmh/

https://www.parleys.com/tutorial/java-microbenchmark-harness-the-lesser-two-evils

Aleksey Shipilëv @shipilev

Чувак из ТВ-службы пришёл отключать антенну. Оказался масс-спектрометристом, сцепился языком с тестем: стоят, обсуждают девайсы. #наукоград

JMHJMH

• Simple à utiliser<dependency>

<groupId>org.openjdk.jmh</groupId><artifactId>jmh‐core</artifactId><version>1.7</version>

</dependency>

JMHJMH

• État partagé entre les exécutions@State(Scope.Benchmark)public class ShakespearePlaysScrabble {

Set<String> scrabbleWords = null ;Set<String> shakespeareWords = null ;

@Setuppublic void init() {

scrabbleWords = Util.readScrabbleWords() ;shakespeareWords = Util.readShakespeareWords() ;

JMHJMH

• Simple à mettre en œuvre @Benchmark@BenchmarkMode(Mode.AverageTime)@OutputTimeUnit(TimeUnit.MILLISECONDS)@Warmup(iterations=5)@Measurement(iterations=5)@Fork(3)public List<Entry<Integer, List<String>>> measureAverage() {

// implementation to test}

JMHJMH

• Utilisation particulière> mvn clean install> java –jar target/benchmark.jar

JMHJMH

• 3 façons de mesurer les performances Temps moyen d’exécution Nombre d’exécutions par seconde Échantillonnage (diagramme par quantiles)

• Mesure du temps moyenBenchmark Mode Cnt Score Error UnitsGSCollections avgt 100 25,392 ± 0,253 ms/op

• Mesure du temps moyenBenchmark Mode Cnt Score Error UnitsGSCollections avgt 100 25,392 ± 0,253 ms/opNonParallelStreams avgt 100 29,027 ± 0,279 ms/op

• Mesure du temps moyenBenchmark Mode Cnt Score Error UnitsGSCollections avgt 100 25,392 ± 0,253 ms/opNonParallelStreams avgt 100 29,027 ± 0,279 ms/opRxJava avgt 100 253,788 ± 1,421 ms/op

• Mesure du temps moyenBenchmark Mode Cnt Score Error UnitsGSCollections avgt 100 25,392 ± 0,253 ms/opNonParallelStreams avgt 100 29,027 ± 0,279 ms/opRxJava avgt 100 253,788 ± 1,421 ms/opParallelStreams avgt 100 7,624 ± 0,055 ms/op

• Mesure du temps moyen

• Attention à la mémoire avec GS Collections

Benchmark Mode Cnt Score Error UnitsGSCollections avgt 100 25,392 ± 0,253 ms/opNonParallelStreams avgt 100 29,027 ± 0,279 ms/opRxJava avgt 100 253,788 ± 1,421 ms/opParallelStreams avgt 100 7,624 ± 0,055 ms/op

Conclusion

Conclusion Conclusion

• La programmation fonctionnelle est « à la mode »• Du point de vue performance, les choses ne sont pas

si simples• Le choix de l’API Stream : « une dose de fonctionnel »

est probablement le bon

• GS Collections : bonne API Beaucoup de patterns en doublon des Streams Compatible Java 7

• RxJava : API riche et complexe Approche différente, patterns complémentaires Attention aux performances

• Java 8 Stream Performante, efficace en mémoire Seule à offrir la parallélisation « gratuite » Extensible au travers des Spliterator

@JosePaumard#J8Stream

Les Streams sont parmi nous

Education

À ma maman, Maguy Guillou, tu seras toujours parmi nous.ekladata.com/...Tome_1_Always_and_fo_-_Faustine_M.pdf · À ma maman, Maguy Guillou, tu seras toujours parmi nous. Ceci est

ATELIER INTERNATIONAL FAIRE MONDE COMMUNpaysages-territoires-transitions.cerema.fr/IMG/pdf/... · Qui sommes-nous parmi les autres êtres vivants ? Quel monde voulons-nous construire

De Sède, Gérard, ‘Les Templiers sont parmi nous’€¦ · De Sède, Gérard, ‘Les Templiers sont parmi nous’ (1962) Vertaling titel: De Tempeliers zijn onder ons (1962) Deel

Introduction programmation Javaairiau/Teaching/L3-Java/...L’idée avec les streams sera d’écrire le code suivant : long count = words.stream().filter(w-> w.length > 12).count();

Hiver, tu es arrivé parmi nous Avec tes fardeaux dévènements Lautomne saison triste aux arbres roux Nous quittera sans regret, et très lentement. La

ILS SONT PARMI NOUS. KABALLAH, ELOHIM, ANGES, DÉMONS ET EXTRATERRESTRESnouvelletorah.com/.../07/Extraterrestres-NouvelleTorah.pdf · 2020. 7. 15. · ILS SONT PARMI NOUS. KABALLAH,

· Web viewPourrions-nous vivre si, en venant au monde, nous n'avions pas droit à un espace sur terre ? Ceux qui parmi nous en font la douloureuse expérience, les exclus, les parias,

Les manipulateurs sont parmi nous

Nous pensons que tous parmi vous ont déjà entendu parler de la … · 2012-12-01 · Nous pensons que tous parmi vous ont déjà entendu parler de la CP5. 1 Rappelons qu’elle

Design Ilot des combes 2015 - Mouvement Colibris pouce» de Colibris, parmi lesquels l’Ilot des Combes, au Creusot. Ce projet nous est apparu Ce projet nous est apparu particulièrement

Integrating Sentiment Analysis and Term …...Integrating Sentiment Analysis and Term Associations with Geo-Temporal Visualizations on Customer Feedback Streams Ming Hao 1, Christian

[infographie] les voitures connectées sont déjà parmi nous

PRELIMINARY INVESTIGATIONS OF DISEASE-CAUSING …€¦ · for ectocommensal peritrich protozoans Epystilis spp. and Cothurnia sieboldii. In crayfish sampled from five streams and

Japanese Youth: An Interactive Dialogue: Towards Comparative … · begin bridging two often disconnected (but potentially mutually enriching) streams of literature, namely, research

Oracle Streams Concepts and Administration - Oracle Documentation

Bioaccumulation d'aluminium chez la truite Salmo truffa ... · Aluminium bioaccumulation in the brown trout Salmo truffa fario from acidified streams: a structural, ultrastructural

CAVAC DE LA MONTÉRÉGIE · de Sylvie Tardif et nous avons la chance qu’elle soit encore parmi nous ! À nos débuts, en 1990, c’était des bénévoles, chapeautés par un coordonnateur

Visa Or Voyages...La carte qui vous fait voyager en toute assurance! Nous sommes heureux de vous compter parmi notre clientèle détentrice d’une carte Visa* Or Voyages. Nous vous

1.Les Manipulateurs Sont Parmi Nous Cle0cbaaf

Schwaarze Mann - Un Noir parmi nous - ZPB