Programmation réactive en Java : la bibliothèque Reactor

Programmation réactive en Java avec Project Reactor

\n\n

La programmation réactive est un paradigme de développement axé sur les flux de données asynchrones et la propagation des changements. En Java, ce concept est implémenté via la spécification Reactive Streams et son implémentation la plus populaire — la bibliothèque Project Reactor (partie de l'écosystème Spring WebFlux).

\n\n

Contrairement au code impératif traditionnel, où le programme exécute séquentiellement des commandes, l'approche réactive permet de traiter les données au fur et à mesure qu'elles arrivent, sans bloquer les threads d'exécution. C'est particulièrement important pour les systèmes à forte charge, les microservices et les applications avec un grand nombre de requêtes simultanées.

\n\n

Composants principaux de Reactor

\n\n

La bibliothèque fournit deux types principaux de séquences réactives :

\n\n

\n
• Mono<T> — représente un flux qui peut contenir 0 ou 1 élément. Utilisé pour les opérations asynchrones unitaires (par exemple, une requête HTTP, la récupération d'un enregistrement en base de données).
\n
• Flux<T> — représente un flux qui peut contenir de 0 à N éléments. Adapté aux collections, aux flux d'événements ou aux données arrivant par parties.
\n

\n\n

Les deux types implémentent l'interface Publisher de la spécification Reactive Streams et supportent le mécanisme de backpressure (contre-pression) — la capacité de l'abonné à contrôler la vitesse de réception des données.

\n\n

Exemple : création et traitement d'un Flux

\n\n

Considérons un exemple simple de création d'un flux réactif de nombres et de son traitement :

\n\n

import reactor.core.publisher.Flux;\n\npublic class ReactorExample {\n    public static void main(String[] args) {\n        Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5);\n\n        numbers\n            .map(n -> n * 2)          // transformation\n            .filter(n -> n % 3 == 0)   // filtrage\n            .subscribe(\n                data -> System.out.println("Reçu : " + data),\n                error -> System.err.println("Erreur : " + error),\n                () -> System.out.println("Terminé !")\n            );\n    }\n}\n

\n\n

Dans cet exemple :

\n

\n
• Flux.just() crée un flux de cinq nombres.
\n
• map() multiplie chaque élément par 2.
\n
• filter() ne conserve que les nombres multiples de 3.
\n
• subscribe() lance le traitement — affiche le résultat, gère les erreurs et signale la fin.
\n

\n\n

Asynchronisme et travail avec Mono

\n\n

Reactor permet d'effectuer des opérations asynchrones sans bloquer les threads. Par exemple, la simulation d'une requête vers un service distant :

\n\n

import reactor.core.publisher.Mono;\nimport java.time.Duration;\n\npublic class AsyncExample {\n    public static void main(String[] args) throws InterruptedException {\n        Mono<String> asyncResult = Mono.fromSupplier(() -> {\n            try {\n                Thread.sleep(1000); // simulation d'un délai\n            } catch (InterruptedException e) {\n                e.printStackTrace();\n            }\n            return "Données reçues";\n        }).subscribeOn(Schedulers.boundedElastic()); // exécution dans un autre thread\n\n        System.out.println("Requête envoyée...");\n        \n        asyncResult.subscribe(System.out::println);\n        \n        Thread.sleep(2000); // attendre la fin\n    }\n}\n

\n\n

Point clé : l'appel .subscribeOn() déplace l'exécution de l'opération coûteuse dans un thread séparé, sans bloquer le thread principal. L'abonné recevra le résultat de manière asynchrone.

\n\n

Gestion des erreurs dans les flux réactifs

\n\n

Contrairement à try-catch, en programmation réactive, les erreurs sont traitées via des opérateurs :

\n\n

Flux<String> fluxWithError = Flux.just("A", "B", "C")\n    .map(s -> {\n        if ("B".equals(s)) t