Contactez-nous

Project Reactor : `Mono` et `Flux`

Plongez au coeur de Project Reactor et découvrez les types fondamentaux Mono (0..1) et Flux (0..N) pour la programmation réactive asynchrone dans l'écosystème Spring.

Introduction à Project Reactor

Project Reactor est une bibliothèque Java de quatrième génération pour la programmation réactive, développée par l'équipe de Spring chez VMware. Elle implémente la spécification Reactive Streams, une initiative standardisant la programmation asynchrone avec gestion de la contre-pression (backpressure) sur la JVM. Reactor est le coeur de la pile réactive de Spring, notamment de Spring WebFlux et de Spring Data Reactive.

Plutôt que de manipuler directement des threads, des `Future` ou des callbacks complexes, Reactor propose un modèle de composition déclaratif pour gérer des flux de données asynchrones. Il fournit deux types de 'publishers' (éditeurs) principaux, conformes à la spécification Reactive Streams, qui sont les briques de base de toute application réactive construite avec Reactor : `Flux` et `Mono`.

Comprendre `Mono` et `Flux` est absolument essentiel pour travailler avec Spring WebFlux ou toute autre partie réactive de l'écosystème Spring. Ils représentent la manière dont les données circulent de manière asynchrone et non bloquante dans ces systèmes.

Flux : La séquence asynchrone (0..N éléments)

Un `Flux` représente un flux asynchrone de 0 à N éléments de type `T`. Il peut émettre zéro, un, plusieurs, voire une infinité d'éléments, suivis éventuellement par un signal de complétion (`onComplete`) ou un signal d'erreur (`onError`). Pensez à un `Flux` comme l'équivalent réactif d'une collection (`List`, `Set`) ou d'un `Stream` Java 8, mais dont les éléments arrivent de manière asynchrone au fil du temps.

Il est idéal pour représenter des ensembles de données dont la taille n'est pas connue à l'avance ou qui arrivent progressivement : résultats d'une requête en base de données retournant plusieurs lignes, messages provenant d'un broker, événements en temps réel, etc.

Voici quelques manières courantes de créer un `Flux` :

import reactor.core.publisher.Flux;
import java.util.List;
import java.time.Duration;

// Créer un Flux à partir d'éléments fixes
Flux fluxJust = Flux.just("Pomme", "Banane", "Orange");

// Créer un Flux à partir d'une collection (Iterable)
List fruitsList = List.of("Fraise", "Framboise");
Flux fluxFromIterable = Flux.fromIterable(fruitsList);

// Créer un Flux à partir d'une plage de nombres
Flux fluxRange = Flux.range(1, 5); // Emet 1, 2, 3, 4, 5

// Créer un Flux qui émet des éléments à intervalle régulier
Flux fluxInterval = Flux.interval(Duration.ofSeconds(1)); // Emet 0, 1, 2, ... chaque seconde

// Créer un Flux vide
Flux fluxEmpty = Flux.empty();

// Créer un Flux qui émet immédiatement une erreur
Flux fluxError = Flux.error(new RuntimeException("Oups !"));

Un `Flux` implémente l'interface `Publisher` de Reactive Streams et supporte la contre-pression (backpressure), permettant au 'Subscriber' (abonné) de contrôler la vitesse à laquelle le `Flux` émet les éléments pour éviter d'être submergé.
Mono : La valeur unique ou l'absence de valeur (0..1 élément)
Un `Mono` représente un flux asynchrone contenant au maximum 0 ou 1 élément de type `T`. C'est une spécialisation de `Flux` optimisée pour les cas où le résultat attendu est une seule valeur ou simplement un signal de complétion sans valeur, ou une erreur.
Il est parfait pour modéliser des opérations asynchrones qui retournent un seul résultat ou rien du tout : une requête en base de données par ID, un appel API retournant un objet unique, une opération de sauvegarde qui réussit ou échoue (représenté par un `Mono`), etc. C'est l'équivalent réactif d'un `Optional` ou d'un `CompletableFuture` qui peut retourner une valeur, être vide, ou échouer.
Voici quelques manières courantes de créer un `Mono` :
import reactor.core.publisher.Mono;
import java.util.Optional;

// Créer un Mono à partir d'une valeur existante (non nulle)
Mono monoJust = Mono.just("Bonjour");

// Créer un Mono vide (signal onComplete sans valeur)
Mono monoEmpty = Mono.empty();

// Créer un Mono qui émet immédiatement une erreur
Mono monoError = Mono.error(new IllegalArgumentException("Paramètre invalide"));

// Créer un Mono à partir d'un Optional
Optional optionalValue = Optional.of("Valeur optionnelle");
Mono monoFromOptional = Mono.justOrEmpty(optionalValue); // Emet "Valeur optionnelle"
Mono monoFromEmptyOptional = Mono.justOrEmpty(Optional.empty()); // Devient un Mono.empty()

// Créer un Mono à partir d'une tâche synchrone (Callable)
// L'exécution est différée jusqu'à la souscription
Mono monoFromCallable = Mono.fromCallable(() -> {
    // Simule une opération bloquante ou coûteuse
    Thread.sleep(100);
    return "Résultat calculé";
});

// Créer un Mono représentant une complétion sans valeur (équivalent à onComplete)
Mono monoFromRunnable = Mono.fromRunnable(() -> {
    System.out.println("Tâche terminée !");
});

Un `Mono` implémente également `Publisher`. Bien qu'il n'émette qu'un seul élément, le concept de souscription et de signal (`onNext`, `onComplete`, `onError`) reste le même.
Caractéristiques clés et opérateurs
Asynchrones et non bloquants : `Mono` et `Flux` représentent des calculs qui produiront des résultats *dans le futur*. Les opérations effectuées sur eux ne bloquent généralement pas le thread appelant.
Immutables : Les opérateurs appliqués à un `Mono` ou `Flux` (comme `map`, `filter`, `flatMap`) ne modifient pas l'instance originale mais retournent une *nouvelle* instance représentant la transformation.
Lazy (Paresseux) : Rien ne se passe tant qu'il n'y a pas de souscription. C'est l'appel à la méthode `subscribe(...)` qui déclenche réellement l'exécution du flux de données et l'application des opérateurs.
Opérateurs : Reactor fournit une API riche et fluide avec des centaines d'opérateurs pour composer, transformer, filtrer et combiner les `Mono` et `Flux`. Quelques exemples fondamentaux :
  • `map(Function)` : Transforme chaque élément de manière synchrone (1-to-1).
  • `flatMap(Function>)` : Transforme chaque élément en un nouveau `Publisher` (`Mono` ou `Flux`) de manière asynchrone, puis 'aplatit' les résultats dans un seul flux. Essentiel pour enchaîner des opérations asynchrones.
  • `filter(Predicate)` : Ne laisse passer que les éléments qui satisfont une condition.
  • `zipWith(Publisher)` : Combine les éléments de deux publishers en paires.
  • `mergeWith(Publisher)` : Fusionne les éléments de deux publishers au fur et à mesure qu'ils arrivent.
  • `doOnNext(Consumer)` / `doOnError(Consumer)` / `doOnComplete(Runnable)` : Permettent d'exécuter des effets de bord (logging, métriques) lors de certains événements du cycle de vie, sans affecter le flux principal.
  • `subscribe(...)` : La méthode terminale qui démarre le flux. Elle peut prendre des consommateurs pour les éléments (`onNext`), les erreurs (`onError`), et la complétion (`onComplete`).
Exemple d'enchaînement simple :
Flux.range(1, 10)
    .filter(i -> i % 2 == 0) // Garde les nombres pairs
    .map(i -> "Nombre: " + i) // Transforme en chaîne
    .subscribe(
        System.out::println, // Action pour chaque élément (onNext)
        error -> System.err.println("Erreur: " + error), // Action en cas d'erreur (onError)
        () -> System.out.println("Flux terminé !") // Action à la complétion (onComplete)
    );
// Output:
// Nombre: 2
// Nombre: 4
// Nombre: 6
// Nombre: 8
// Nombre: 10
// Flux terminé !

Rôle dans Spring Boot et WebFlux
Dans le contexte de Spring Boot et plus spécifiquement de Spring WebFlux, `Mono` et `Flux` sont omniprésents :
  • Contrôleurs WebFlux : Les méthodes des contrôleurs retournent typiquement `Mono>`, `Mono`, ou `Flux` pour indiquer une réponse asynchrone.
  • Clients Web réactifs (`WebClient`) : Les appels HTTP effectués avec `WebClient` retournent des `Mono` ou des `Flux` représentant la réponse asynchrone.
  • Repositories Réactifs (Spring Data Reactive) : Les méthodes des interfaces de repositories réactifs (ex: `R2dbcRepository`, `ReactiveMongoRepository`) retournent des `Mono` (pour `findById`, `save`) ou des `Flux` (pour `findAll`).
L'utilisation de ces types permet à Spring WebFlux de fonctionner sur un modèle d'exécution non bloquant basé sur une boucle d'événements (event loop), utilisant un petit nombre de threads pour gérer un grand nombre de requêtes concurrentes efficacement. En retournant un `Mono` ou un `Flux`, votre code signale au framework qu'il n'a pas besoin de bloquer un thread en attendant le résultat, libérant ainsi le thread pour traiter d'autres requêtes.
En résumé, `Mono` et `Flux` sont les piliers de la programmation réactive avec Project Reactor. Ils offrent un moyen puissant et déclaratif de gérer des séquences de données asynchrones, qu'elles contiennent zéro, une ou plusieurs valeurs, et constituent la base de la pile réactive moderne de Spring.