API Asynchrones en Java 8

@JosePaumard

asynchrones

Java

#J8Async @JosePaumard

Asynchrone ?


AsynchroneAsynchrone

• Trois tâches à exécuterT1

T2

T3



• 1ère façon de faire :

« exécution synchrone »



• 2ème façon de faire :

« exécution multithread »




« exécution multithread » … sur un seul cœur




« asynchrone »




« asynchrone » … même sur un multicœur




• Plus rapide ?



• Plus rapide ? En général oui Approche « non blocking »



• Différence avec le modèle synchrone multithread ?1) Le traitement décide de passer d’une tâche à l’autre2) Pas de problème d’atomicité / visibilité

• Performances ? Pas de « context switch »



• Pattern


queryEngine.select("select user from User").forEach(user ‐> System.out.prinln(user)) ;


• Pattern

• Callback ou tâche : lambda expression


queryEngine.select("select user from User").forEach(user ‐> System.out.prinln(user)) ;


• Pattern

• Callback ou tâche : lambda expression• Enchaînement : lorsque le résultat est disponible

alors on enchaîne avec le traitement


queryEngine.select("select user from User").forEach(System.out::prinln) ;


• Pattern

• Callback ou tâche : lambda expression• Enchaînement : lorsque le résultat est disponible

alors on enchaîne avec le traitement

• Comment écrire ceci en Java ?


queryEngine.select("select user from User").forEach(System.out::prinln) ;


Notion de tâcheNotion de tâche

• Depuis Java 1 : Runnable• Java 5 : Callable

• Java 5 : ExecutorService (pool de threads)

• On donne une tâche, on récupère un Future



• Pattern Callable<String> task = () ‐> "select user from User" ;Future<String> future = executorService.submit(task) ;



• Pattern Callable<String> task = () ‐> "select user from User" ;Future<String> future = executorService.submit(task) ;

List<User> users = future.get() ; // blockingusers.forEach(System.out::println) ;



• Pattern

• Le passage d’un objet d’une tâche à l’autre se fait dans le thread « maître »

Callable<String> task = () ‐> "select user from User" ;Future<String> future = executorService.submit(task) ;

List<User> users = future.get() ; // blockingusers.forEach(System.out::println) ;


Programmation asynchroneProgrammation asynchrone

• Nouveaux outils en Java 8 pour traiter ce point• Solution pour enchaîner les tâches

• Asynchrone & multithread


Questions ?

#J8Async

Questions ?

#J8Async


• Nouvelle interface en Java 8 : CompletionStage

De quoi s’agit-il ? De quoi s’agit-il ?

/*** A stage of a possibly asynchronous computation, that performs an* action or computes a value when another CompletionStage completes.* A stage completes upon termination of its computation, but this may* in turn trigger other dependent stages.*/


• Nouvelle interface en Java 8 : CompletionStage

• CompletionStage = une tâche qui se déclenche sur une autre et qui peut en déclencher d’autres


/*** A stage of a possibly asynchronous computation, that performs an* action or computes a value when another CompletionStage completes.* A stage completes upon termination of its computation, but this may* in turn trigger other dependent stages*/


• Classe d’implémentation : CompletableFuture

• Implémente à la fois : Future CompletionStage



• CompletableFuture : modélise une tâche• Peut être dans trois états :



• CompletableFuture : modélise une tâche• Peut être dans trois états : En train d’être calculée



• CompletableFuture : modélise une tâche• Peut être dans trois états : En train d’être calculée Calculée, ayant produit un résultat



• CompletableFuture : modélise une tâche• Peut être dans trois états : En train d’être calculée Calculée, ayant produit un résultat Calculée, ayant généré une exception



Future Future

• Cinq méthodes :boolean cancel(boolean mayInterruptIfRunning) ;


Future Future


boolean isCanceled() ;boolean isDone() ;


Future Future


boolean isCanceled() ;boolean isDone() ;

V get() ; // blocking call V get(long timeout, TimeUnit timeUnit) ; // may throw a checked exception

throws InterruptedException, ExecutionException, TimeoutException ;


CompletableFutureCompletableFuture

• Méthodes de type « future » :V join() ; // may throw an unchecked exceptionV getNow(V valueIfAbsent) ; // returns immediately




boolean complete(V value) ; // sets the returned value is not returnedvoid obtrudeValue(V value) ; // resets the returned value




boolean complete(V value) ; // sets the returned value is not returnedvoid obtrudeValue(V value) ; // resets the returned value

boolean completeExceptionnaly(Throwable t) ; // sets an exceptionvoid obtrudeException(Throwable t) ; // resets with an exception


Création d’un CompletableFutureCréation d’un CompletableFuture

• CompletableFuture déjà terminépublic static CompletableFuture completedFuture(U value) ;


Création d’un CompletableFutureCréation d’un CompletableFuture

• CompletableFuture déjà terminépublic static CompletableFuture completedFuture(U value) ;

public static CompletableFuture supplyAsync(Supplier value, Executor executor) ;

public static CompletableFuture runAsync(Runnable runnable, Executor executor) ;


CompletionStageCompletionStage

• Concept : un étape dans un traitement global Peut être déclenchée par une étape précédente Peut déclencher d’autres étapes Peut être exécutée dans un executor particulier


CompletionStageCompletionStage

• Notion tâche : Function : prend un argument, retourne une valeur Consumer : prend un argument Runnable

= interfaces fonctionnelles, donc lambda


CompletionStage – chaînage CompletionStage – chaînage

• Chaînage après, même threadpublic CompletionStage

thenApply(Function<? super T,? extends U> fn);

public CompletionStage<Void> thenAccept(Consumer<? super T> action);

public CompletionStage<Void> thenRun(Runnable action);



• Chaînage après, autre thread (common FJ pool)public CompletionStage

thenApplyAsync(Function<? super T,? extends U> fn);

public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action);

public CompletionStage<Void> thenRunAsync(Runnable action);



• Chaînage après, autre thread (executor)public CompletionStage

thenApplyAsync(Function<? super T,? extends U> fn, Executor executor);

public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor);

public CompletionStage<Void> thenRunAsync(Runnable action, Executor executor);


CompletionStage – compositionCompletionStage – composition

• Composition public CompletionStage

thenCompose(Function<? super T, ? extends CompletionStage> fn);

public CompletionStage<Void> thenComposeAsync(

Function<? super T, ? extends CompletionStage> fn);

public CompletionStage<Void> thenComposeAsync(

Function<? super T, ? extends CompletionStage> fn,Executor executor);


Chaînage & compositionChaînage & composition

• Ces deux familles de fonction permettent d’enchaîner une opération après l’autre



• On peut aussi enchaîner une tâche à la suite de deux autres tâches

public CompletionStage<V> thenCombine(CompletionStage other, BiFunction<T, U, V> function) ;




• Prend les résultats de this et other Et les combine dans function






public CompletionStage<V> thenCombineAsync(CompletionStage other, BiFunction<T, U, V> function) ;





public CompletionStage<V> thenCombineAsync(CompletionStage other, BiFunction<T, U, V> function, Executor executor) ;



• Versions avec Consumerpublic CompletionStage<Void> thenAcceptBoth

(CompletionStage other, BiConsumer<T, U> action) ;

public CompletionStage<Void> thenAcceptBothAsync(CompletionStage other, BiConsumer<T, U> action) ;

public CompletionStage<Void> thenAcceptBothAsync(CompletionStage other, BiConsumer<T, U> action, Executor executor) ;



• Versions avec Runnablepublic CompletionStage<Void> runAfterBoth

(CompletionStage<?> other, Runnable action) ;

public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action) ;

public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action, Executor executor) ;



• Ces tâches se déclenchent conditionnellement à thiset à la tâche passée en paramètre

• Lorsque ces tâches sont terminées

• On peut aussi déclencher lorsque la première se termine


Chaînage multipleChaînage multiple

• Version functionpublic CompletionStage<V> applyToEither

(CompletionStage<? extends T> other, Function<T, U> function) ;



• Version functionpublic CompletionStage applyToEither

(CompletionStage<? extends T> other, Function<T, U> function) ;

public CompletionStage applyToEitherAsync(CompletionStage<? extends T> other, Function<T, U> function) ;

public CompletionStage applyToEitherAsync(CompletionStage<? extends T> other, Function<T, U> function, Executor executor) ;



• Version consumerpublic CompletionStage<V> acceptEither

(CompletionStage<? extends T> other, Consumer<? extends T> consumer) ;

public CompletionStage<V> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? extends T> consumer) ;

public CompletionStage<V> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? extends T> consumer, Executor executor) ;



• Version runnablepublic CompletionStage<V> runAfterEither

(CompletionStage other, Runnable action) ;

public CompletionStage<V> runAfterEitherAsync(CompletionStage other, Runnable action) ;

public CompletionStage<V> runAfterEitherAsync(CompletionStage other, Runnable action, Executor executor) ;


Création sur plusieurs tâchesCréation sur plusieurs tâches

• Méthodes statiquespublic static CompletableFuture<Void>

allOf(CompletableFuture<?>... cfs) ;

public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) ;



• Attention à la sémantique !

• Imprime « null »

public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) ;

CompletableFuture<Void> allOf = CompletableFuture.allOf() ;System.out.println("allOF : " + allOf.join()) ;



• Attention à la sémantique !

• Ne rend pas la main…

public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) ;

CompletableFuture<Object> anyOf = CompletableFuture.anyOf() ;System.out.println("anyOf : " + anyOf.join()) ;



• Attention à la sémantique !public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) ;

CompletableFuture<Object> anyOf = CompletableFuture.anyOf() ;System.out.println("anyOf : " + anyOf.getNow("Nothing to say")) ;


Gestion des exceptionsGestion des exceptions

• Point délicat : Une première étape consiste à créer les tâches et à

décrire leur enchaînement L’exécution des tâches démarre indépendamment des

appels À chaque étape, un CompletableFuture est créé



• Un CompletableFuture peut dépendre : Cas 1 : d’un autre CompletableFuture Cas 2 : de deux autres CompletableFuture Cas 3 : de N CompletableFuture



• Une exception est jetée dans le cas 1 Tous les CompletableFuture sont en erreur• Ils se terminent « exceptionnellement » isExceptionnaly() retourne true L’appel à get() jette une ExecutionException get().getCause() retourne l’exception première



• Une exception est jetée dans le cas 2 Tous les CompletableFuture en aval sont en erreur• Ils se terminent « exceptionnellement »• L’autre tâche peut se terminer normalement On peut l’interroger par get() pour avoir son résultat



• Une exception est jetée dans le cas 3 Le CompletableFuture retourné est en erreur• Il se termine « exceptionnellement »• Les autres tâches peuvent se terminer normalement On peut l’interroger par get() pour avoir son résultat



• On peut aussi traiter une exception normalement

Dans ce cas, l’exception est passée à la fonction Utile pour les checked exception

CompletionStage<T> exceptionally(Function<Throwable, ? extends T> function);



• Méthode whenComplete()

Dans ce cas t ou e est nul dans l’appel de action Le CompletableFuture retourné peut ne pas être en

erreur

CompletionStage<T> whenComplete(BiConsumer<T, Throwable> action) ;



• Méthode whenComplete()CompletionStage<T> whenComplete

(BiConsumer<T, Throwable> action) ;

CompletionStage<T> whenCompleteAsync(BiConsumer<T, Throwable> action) ;

CompletionStage<T> whenCompleteAsync(BiConsumer<T, Throwable> action, Executor executor) ;



• Méthode handle()

Dans ce cas t ou e est nul dans l’appel de function Retourne un CompletableFuture qui peut ne pas être

en erreur

CompletionStage<T> handle(BiFunction<T, Throwable, U> function) ;



• Méthode handle()CompletionStage<T> handle

(BiFunction<T, Throwable, U> function) ;

CompletionStage<T> handleAsync(BiFunction<T, Throwable, U> function) ;

CompletionStage<T> handleAsync(BiFunction<T, Throwable, U> function, Executor executor) ;


Une dernière méthodeUne dernière méthode

• CompletableFuture : On peut obtenir une estimation du nombre de tâches qui attendent l’exécution d’une tâche donnée

int getNumberOfDependents() ;


Exemple – 1 Exemple – 1

• Lecture asynchrone de liens et affichageCompletableFuture.supplyAsync(

() ‐> readPage("http://whatever.com/"))




() ‐> readPage("http://whatever.com/")).thenApply(page ‐> linkParser.getLinks(page))




() ‐> readPage("http://whatever.com/")).thenApply(page ‐> linkParser.getLinks(page)).thenAccept(

links ‐> displayPanel.display(links)

) ;




() ‐> readPage("http://whatever.com/")).thenApply(page ‐> linkParser.getLinks(page)).thenAcceptAsync(

links ‐> displayPanel.display(links),executor

) ;



• Lecture asynchrone de liens et affichagepublic interface Executor {

void execute(Runnable command);}



• Lecture asynchrone de liens et affichagepublic interface Executor {

void execute(Runnable command);}

Executor executor = runnable ‐> SwingUtilities.invokeLater(runnable) ;




() ‐> readPage("http://whatever.com/")).thenApply(page ‐> linkParser.getLinks(page)).thenAcceptAsync(

links ‐> displayPanel.display(links), runnable ‐> SwingUtilities.invokeLater(runnable)

) ;




() ‐> readPage("http://whatever.com/")).thenApply(Parser::getLinks).thenAcceptAsync(

DisplayPanel::display, SwingUtilities::invokeLater

) ;



• Événements asynchrones dans CDI@InjectEvent<String> event ;

event.fire("some event") ; // returns void

public void observes(@Observes String payload) {// handle the event, called in the firing thread

}



• Événements asynchrones dans CDIpublic void observes(@Observes String payload) {

// handle the event, called in the firing threadCompletableFuture.anyOf(/* some task */) ;

}




event.fireAsync("some event") ; // returns CompletableFuture<Object> (?)

public void observes(@Observes String payload) {// handle the event in another thread

}




event.fireAsync("some event", executor) ;

public void observes(@Observes String payload) {// handle the event in the executor

}





@Produces @SwingExecutorExecutor executor = SwingUtilities::invokeLater

public void observes(@Observes String payload, @SwingExecutor Executor executor) {

// handle the event in the Swing thread}





@Produces @SwingExecutorExecutor executor = SwingUtilities::invokeLater

public void observes(@Observes @SwingExecutor String payload) {// handle the event in the Swing thread

}



CompletableFuture<String> closing = new CompletableFuture<String>() ;Stream<String> manyStrings = Stream.of("one", "two", "three") ;

CompletableFuture<String> reduce =manyStrings

.onClose(() ‐> { closing.complete("Closed") ; })

.map(CompletableFuture::completedFuture)

.reduce(closing, (cf1, cf2) ‐> cf1.thenCombine(cf2, function) // concatenation

) ;manyStrings.close() ;


L’indispensable !L’indispensable !

• Fixer la taille du Common Fork / Join PoolSystem.setProperty(

"java.util.concurrent.ForkJoinPool.common.parallelism", 2) ;


Conclusion Conclusion

• On a une API pour le calcul asynchrone dans le JDK !



• On a une API pour le calcul asynchrone dans le JDK !• Très riche et souple à l’utilisation



• On a une API pour le calcul asynchrone dans le JDK !• Très riche et souple à l’utilisation• Construite sur l’utilisation des lambda



• On a une API pour le calcul asynchrone dans le JDK !• Très riche et souple à l’utilisation• Construite sur l’utilisation des lambda• Permet un contrôle fin des threads



• On a une API pour le calcul asynchrone dans le JDK !• Très riche et souple à l’utilisation• Construite sur l’utilisation des lambda• Permet un contrôle fin des threads• Gère différents types de chaînage



• On a une API pour le calcul asynchrone dans le JDK !• Très riche et souple à l’utilisation• Construite sur l’utilisation des lambda• Permet un contrôle fin des threads• Gère différents types de chaînage• Gère intelligemment les exceptions

@JosePaumard#J8Stream

@JosePaumard#J8Stream

Education

API Asynchrones en Java 8