RxJava : bases, ReactiveX et travail avec les flux de données

Qu'est-ce que RxJava et ReactiveX

RxJava est l'implémentation Java de la spécification ReactiveX, une bibliothèque pour la programmation asynchrone utilisant des flux observables (Observable). RxJava 2 a été publié en 2016 avec la prise en charge de Reactive Streams (Flowable) et la séparation en rx.Observable et io.reactivex.Observable. RxJava 3 (2019) est la version majeure actuelle avec une compatibilité ascendante avec RxJava 2.

L'idée centrale de RxJava est que tout est un flux : flux de données, flux d'événements, flux d'états. Toute opération asynchrone peut être représentée comme un Observable émettant des données, une erreur ou un signal d'achèvement. Un Observer s'abonne à l'Observable et reçoit des notifications en temps réel.

Selon Badoo (2024), avant la transition vers les coroutines, 76 % des applications Android du top 200 du Google Play utilisaient RxJava pour les opérations asynchrones. La part diminue maintenant en faveur des coroutines, mais RxJava reste dans le code de production de milliers d'applications et est considéré comme une technologie mature et éprouvée. ReactiveX est une spécification multiplateforme également implémentée pour JavaScript (RxJS), .NET (Rx.NET), Swift (RxSwift) et d'autres langages.

Modèle Observer dans RxJava

ReactiveX étend le modèle Observer classique avec deux mécanismes : le chaînage d'opérateurs et la gestion des threads basée sur les Schedulers. Observable ne commence à émettre des données que lorsqu'un Observer s'abonne (évaluation paresseuse). Cela permet de construire un pipeline de données qui s'active uniquement lorsqu'un abonnement existe.

Types d'Observable : Observable, Flowable, Single, Maybe, Completable

Observable — le type de base émettant 0..N éléments avec onError ou onComplete. Convient pour les flux de données illimités — par exemple, les événements de clic ou les mises à jour de géolocalisation. Observable ne prend pas en charge le backpressure.

Flowable — la version Reactive Streams d'Observable avec prise en charge du backpressure. Utilisé lorsque la source de données peut générer des éléments plus rapidement que l'Observer ne peut les traiter. Flowable prend en charge les stratégies BACKPRESSURE_BUFFER, DROP, LATEST et ERROR.

Single, Maybe et Completable

Single émet exactement un élément ou une erreur — idéal pour les requêtes réseau. Maybe émet 0 ou 1 élément, adapté au cache où les données peuvent être absentes. Completable émet seulement onComplete ou onError, sans données, pratique pour les opérations d'écriture ou de suppression. Ces types simplifient l'API en restreignant le contrat à un cas spécifique. Retrofit (un client HTTP populaire pour Android) prend en charge directement les cinq types RxJava, permettant de choisir le type de retour le plus approprié pour chaque point de terminaison sans code répétitif supplémentaire.

Opérateurs RxJava : transformation et filtrage des flux

Les opérateurs sont des fonctions qui transforment un Observable en un autre. Une chaîne d'opérateurs décrit le pipeline de données : chaque opérateur prend le flux du précédent, le transforme et le transmet au suivant. RxJava contient plus de 200 opérateurs regroupés en catégories.

• map — transforme chaque élément (Integer → String)
• flatMap — transforme un élément en Observable et fusionne le tout en un seul flux
• filter — filtre les éléments qui satisfont une condition
• zip — combine les éléments de N Observables par index
• merge — fusionne plusieurs Observables en un seul, en préservant l'ordre chronologique
• debounce — émet des éléments seulement si un intervalle de temps spécifié s'est écoulé sans autre émission

flatMap est l'un des opérateurs les plus puissants de RxJava. Il permet d'exécuter une requête asynchrone pour chaque élément et de collecter les résultats dans un flux commun. Par exemple, flatMap est utilisé pour charger les détails à partir d'une liste d'ID : chaque ID → requête réseau → fusion des résultats. Contrairement à map, qui transforme simplement un élément, flatMap peut émettre plusieurs éléments ou passer à un autre Observable, ce qui en fait la base pour la construction de pipelines asynchrones.

Gestion des erreurs avec les opérateurs

onErrorResumeNext — bascule vers un Observable de secours en cas d'erreur. retry — se réabonne N fois en cas d'erreur. onErrorReturn — retourne une valeur par défaut au lieu de l'erreur. doOnError — exécute un effet secondaire en cas d'erreur sans modifier le flux (journalisation ou analyse). La combinaison de ces opérateurs permet de construire des pipelines robustes avec une stratégie de gestion des erreurs claire sans try/catch manuel.

Schedulers : gestion des threads dans RxJava

Les Schedulers déterminent sur quel thread Observable et Observer s'exécutent. subscribeOn définit le thread pour la source, observeOn définit le thread pour l'Observer et les opérateurs suivants. Cette séparation est un avantage clé de RxJava : source sur le thread IO, traitement sur computation, UI sur le thread principal.

Principaux Schedulers : Schedulers.io() — pour les opérations d'E/S (réseau, disque), pool illimité. Schedulers.computation() — pour les calculs, pool fixe dimensionné en fonction du nombre de cœurs. Schedulers.newThread() — un nouveau thread pour chaque tâche. AndroidSchedulers.mainThread() — thread principal Android (RxAndroid). Il existe également Schedulers.trampoline() pour exécuter des tâches dans le thread actuel avec une file d'attente FIFO, utile pour les tests.

Selon Google (2025), l'utilisation correcte des Schedulers est la partie la plus difficile de RxJava pour les débutants. Une erreur typique consiste à appeler subscribeOn après observeOn, ce qui n'affecte pas la source. subscribeOn doit être le premier dans la chaîne pour la source, observeOn avant l'abonnement UI. Règle : subscribeOn n'affecte que l'amont (source), observeOn bascule l'aval (abonné et tous les opérateurs après lui).

Exemples de code RxJava dans Android

Considérons trois scénarios : une requête réseau avec Single, des requêtes parallèles avec zip et un debounce pour un champ de recherche avec debounce.

Requête réseau avec Single

Single est parfait pour les requêtes Retrofit : une requête — une réponse. Abonnez-vous sur le thread principal pour les mises à jour de l'interface utilisateur.

api.getUser(id)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new SingleObserver<User>() {
        @Override
        public void onSuccess(User user) { showUser(user); }
        @Override
        public void onError(Throwable e) { showError(e); }
    })

Requêtes parallèles avec zip

zip combine les résultats de deux Singles indépendants en un seul. Ils s'exécutent en parallèle, le résultat est produit après l'achèvement des deux.

Single.zip(
    api.getProfile(),
    api.getSettings(),
    (profile, settings) -> new Dashboard(profile, settings)
)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(dashboard -> showDashboard(dashboard), e -> logError(e))

Debounce pour champ de recherche

debounce ignore les changements rapides de texte et envoie une requête seulement après une pause de 400 ms. distinctUntilChanged annule la requête si le texte n'a pas changé.

RxTextView.textChanges(searchView)
    .debounce(400, TimeUnit.MILLISECONDS)
    .filter(text -> text.length() >= 3)
    .distinctUntilChanged()
    .switchMap(query -> api.search(query))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(results -> showResults(results))

RxJava vs Kotlin Coroutines : comparaison des approches

RxJava et Kotlin Coroutines résolvent le même problème — la programmation asynchrone — mais avec des approches fondamentalement différentes. RxJava est construit sur le modèle Observer et est basé sur le push : la source envoie des données, l'Observer réagit. Les coroutines sont basées sur le pull : le code demande des données séquentiellement via await.

• RxJava — réactif, flux de données, >200 opérateurs, push-based, courbe d'apprentissage abrupte
• Coroutines — séquentiel, suspend/await, ~40 fonctions, pull-based, syntaxe simple
• RxJava — mature (2016), écosystème énorme, mais courbe d'apprentissage abrupte
• Coroutines — moderne (2018), choix préféré de Google pour le nouveau code
• RxJava — backpressure intégré via Flowable, stratégies de tampon bien testées
• Coroutines — Flow avec backpressure est récent, mais activement développé par JetBrains

Selon Google I/O 2024, Kotlin Coroutines est l'approche recommandée pour le nouveau code asynchrone dans Android. RxJava reste pris en charge pour les projets existants. Google fournit des bibliothèques de pont (kotlinx-coroutines-rx3) pour une migration progressive. AndroidX (LiveData, Room, Paging 3) prend en charge les deux approches, permettant d'utiliser RxJava dans les anciens modules et les coroutines dans les nouveaux sans conflits de dépendances.

Stratégie de migration de RxJava vers les coroutines

Transition progressive : chaque nouveau composant est écrit avec des coroutines, l'ancien code RxJava n'est pas modifié. RxJava → coroutines via awaitSingle() ou awaitFirst(). Coroutines → RxJava via future() ou asFlowable(). La migration complète prend de 6 à 18 mois pour les grands projets.