RxJava pour Java : guide complet de la programmation réactive

1. Qu'est-ce que RxJava et pourquoi l'utiliser ?

RxJava est une bibliothèque de programmation réactive pour le langage Java, implémentant le concept d'Observable (observé) et d'Observer (observateur). Elle permet de traiter des flux de données asynchrones et des événements à l'aide d'opérateurs composables. Contrairement aux callbacks traditionnels ou aux Future/Promise, RxJava offre un ensemble puissant d'outils pour transformer, filtrer et combiner des flux de données, rendant le code plus lisible, prévisible et moins sujet aux erreurs liées au multithreading.

Pourquoi utiliser RxJava ? Dans le développement moderne d'applications Java (notamment Android, applications serveur avec Spring WebFlux, microservices), les tâches asynchrones sont omniprésentes : chargement de données réseau, interactions avec des bases de données, gestion des événements utilisateur. RxJava répond à ces défis en fournissant une API unifiée pour travailler avec tout type de séquence asynchrone. Elle permet d'éviter le "callback hell" (l'enfer des callbacks), simplifie la gestion des erreurs dans les chaînes asynchrones et offre des outils puissants pour la gestion des threads (Schedulers).

Les principaux avantages de RxJava : la déclarativité (vous décrivez ce qui doit être fait, pas comment), la composabilité (les opérateurs se combinent facilement), la gestion de la backpressure (contre-pression) pour contrôler la vitesse de consommation des données, et un écosystème riche (RxAndroid, RxJava2Extensions, RxRelay). La bibliothèque est activement maintenue et largement utilisée dans l'industrie, en particulier dans le développement Android (Retrofit + RxJava, Room + RxJava).

2. Installation

RxJava est distribuée via Maven Central. Pour l'ajouter à votre projet, incluez la dépendance dans votre fichier build.gradle (pour Gradle) ou pom.xml (pour Maven).

Gradle (projet Android / Java)

dependencies {
    implementation 'io.reactivex.rxjava3:rxjava:3.1.8'
}

Maven

<dependency>
    <groupId>io.reactivex.rxjava3</groupId>
    <artifactId>rxjava</artifactId>
    <version>3.1.8</version>
</dependency>

Pour les projets Android, on ajoute souvent RxAndroid pour travailler avec le thread principal :

implementation 'io.reactivex.rxjava3:rxandroid:3.0.2'

3. Démarrage rapide — exemple minimal fonctionnel

Créons un Observable simple qui émet les chaînes "Bonjour" et "Monde", puis abonnons-nous à lui :

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;

public class RxJavaQuickStart {
    public static void main(String[] args) {
        // Création d'un Observable qui émet deux éléments
        Observable<String> observable = Observable.just("Bonjour", "Monde");

        // Création d'un Observer
        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("Abonnement établi");
            }

            @Override
            public void onNext(String s) {
                System.out.println("Reçu : " + s);
            }

            @Override
            public void onError(Throwable e) {
                System.err.println("Erreur : " + e.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("Flux terminé");
            }
        };

        // Abonnement
        observable.subscribe(observer);
    }
}

Sortie :

Abonnement établi
Reçu : Bonjour
Reçu : Monde
Flux terminé

4. Méthodes/classes principales

4.1 Observable.just()

Signature : public