RxJava : essence, composants et programmation réactive

Qu'est-ce que RxJava ?

RxJava est une implémentation de la bibliothèque ReactiveX (Reactive Extensions) pour la Machine Virtuelle Java. La première version de RxJava a été publiée par Netflix en 2013 pour gérer les appels asynchrones dans les applications serveur. Au moment de sa création, les principales alternatives en Java étaient Future et Callback — les deux approches menaient à l'enfer des callbacks et à une gestion complexe des threads. RxJava a introduit la composition d'opérations asynchrones via Observable avec des chaînes d'opérateurs fonctionnels.

L'architecture de RxJava est basée sur la spécification Reactive Streams — un standard pour le traitement asynchrone de flux avec contre-pression non bloquante. La spécification définit quatre interfaces : Publisher, Subscriber, Subscription et Processor. RxJava 2+ implémente complètement Reactive Streams via le type Flowable, en respectant les contrats de contre-pression contrairement à RxJava 1. Observable dans RxJava 2 ne supporte pas la contre-pression — il est destiné aux flux avec un petit nombre d'événements ou aux événements UI.

Selon l'enquête JetBrains, 2025, RxJava fait partie des 3 bibliothèques principales pour le développement Android. Les principaux cas d'utilisation incluent : la gestion des requêtes réseau via Retrofit (intégré avec RxJava via CallAdapter), le travail avec Room (les requêtes réactives retournent Flowable ou Maybe), les animations et événements UI via RxBinding, et la recherche avec debounce à la saisie de texte. Tous ces scénarios partagent un modèle de chaîne commun : source (Observable) → transformation (opérateurs) → abonnement (subscribe).

Historique des versions de RxJava

RxJava 1 (2013) a posé les bases avec Observable et les opérateurs, mais souffrait de problèmes de contre-pression — dans les flux rapides, les données s'accumulaient en mémoire, provoquant OutOfMemoryError. RxJava 2 (2016) a corrigé l'architecture en séparant Observable (sans contre-pression) et Flowable (avec contre-pression). RxJava 3 (2020) a ajouté le support de Java 8 Stream API, des opérateurs supplémentaires et des performances d'abonnement améliorées. Actuellement, RxJava 3 est la version recommandée pour les nouveaux projets.

Types de flux réactifs dans RxJava

RxJava fournit cinq types principaux de sources réactives, chacun conçu pour un scénario spécifique. Observable et Flowable émettent plusieurs valeurs, Single émet une valeur ou une erreur, Completable émet seulement l'achèvement sans données, et Maybe émet une valeur, zéro ou une erreur. Choisir le bon type réduit le volume de code et rend la chaîne auto-documentée.

Flowable est le type le plus flexible pour travailler avec de grands flux de données. Il implémente le Publisher de Reactive Streams avec support de la contre-pression : le consommateur peut demander un nombre spécifique d'éléments via Subscription.request(n). Cela évite le débordement du tampon lorsque les vitesses du producteur et du consommateur ne correspondent pas. Si la contre-pression n'est pas critique, utilisez Observable — il a moins de surcharge due à l'absence du mécanisme de request.

Single est le choix optimal pour les requêtes HTTP. Retrofit 2 avec RxJava CallAdapter retourne Single<ResponseBody> pour chaque requête. Single garantit exactement un appel à onSuccess ou onError, ce qui correspond à la sémantique d'une requête HTTP — une réponse ou une erreur. Completable est utilisé pour les opérations d'écriture qui ne retournent pas de données : insert, update, delete. Maybe est pratique pour la vérification de cache — il peut retourner une valeur ou non.

// Exemple d'utilisation de Single pour une requête HTTP
interface ApiService {
    @GET("users/{id}")
    fun getUser(@Path("id") userId: Int): Single<User>
}

// Abonnement avec traitement sur le thread principal
apiService.getUser(42)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ user ->
        textView.text = user.name
    }, { error ->
        Log.e("API", "Error: ${error.message}")
    })
    .addTo(compositeDisposable)

Opérateurs de transformation et de gestion des flux

Les opérateurs dans RxJava sont des fonctions d'ordre supérieur qui prennent une source réactive et en retournent une autre, transformant le flux de données. RxJava 3 contient plus de 400 opérateurs, divisés en catégories : transformation, filtrage, combinaison, gestion des erreurs et gestion du temps. Chaque opérateur est paresseux — la chaîne est construite à la déclaration et exécutée à l'abonnement.

Opérateurs de transformation

map est l'opérateur de base qui transforme chaque valeur via une fonction. flatMap prend une fonction qui retourne un Observable pour chaque élément et aplatit le résultat en un seul flux. switchMap est similaire à flatMap, mais lorsqu'un nouvel élément arrive, il se désabonne de l'Observable précédent. concatMap préserve l'ordre des éléments — contrairement à flatMap, il s'abonne séquentiellement à chaque Observable imbriqué.

// Analyse JSON avec transformation et filtrage
apiService.getUsers()
    .flatMap { users ->
        Observable.fromIterable(users)
    }
    .filter { user ->
        user.age >= 18
    }
    .map { user ->
        UserDto(user.name, user.age)
    }
    .toList()
    .subscribeOn(Schedulers.computation())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ adapter.submitList(it) },
               { Log.e("Erreur", it.message) })

La combinaison de flux est un domaine où RxJava excelle particulièrement. zip combine les éléments de plusieurs Observables par paires selon l'index : premier avec premier, deuxième avec deuxième. combineLatest émet une nouvelle valeur lorsque n'importe quel flux change, combinant les dernières valeurs de tous les flux. merge combine plusieurs Observables en un seul, préservant l'ordre d'arrivée des événements. concat s'abonne séquentiellement à chaque Observable et transmet tous ses événements avant de passer au suivant.

La gestion du temps inclut debounce (attendre une pause dans le flux avant d'émettre), throttleFirst (émettre le premier événement, ignorer les suivants dans une fenêtre), timeout (erreur si aucun événement n'arrive dans l'intervalle). La recherche avec debounce à la saisie de texte est le scénario le plus courant : searchObservable.debounce(300, MILLISECONDS).distinctUntilChanged() évite les requêtes inutiles lors de la saisie rapide.

Schedulers et multithreading

Scheduler dans RxJava est une abstraction sur un pool de threads. La bibliothèque fournit cinq Schedulers intégrés : Schedulers.io() pour les opérations d'E/S (réseau, fichiers), Schedulers.computation() pour les tâches intensives en CPU, Schedulers.newThread() pour un nouveau thread à chaque fois, Schedulers.single() pour l'exécution mono-thread et Schedulers.trampoline() pour l'exécution immédiate dans le thread actuel.

subscribeOn et observeOn

subscribeOn détermine sur quel Scheduler l'Observable source est exécuté. S'il y a plusieurs subscribeOn dans la chaîne, la priorité va au plus proche de la source. observeOn change le downstream vers le Scheduler spécifié — chaque utilisation d'observeOn change le thread pour les opérateurs suivants. Un pattern Android typique : subscribeOn(Schedulers.io()) pour les opérations réseau, observeOn(AndroidSchedulers.mainThread()) pour les mises à jour UI.

// Traitement multithread avec changement de contexte
Observable.fromCallable(() -> database.getItems())
    .subscribeOn(Schedulers.io())            // BD sur io
    .map(items -> processItems(items))     // transformation sur io
    .observeOn(Schedulers.computation())    // passer à computation
    .map(processed -> compressImages(processed))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> ui.showResult(result))

AndroidSchedulers.mainThread() est un Scheduler de la bibliothèque RxAndroid qui exécute du code sur le thread principal d'Android. Il est obligatoire pour toute mise à jour UI dans une chaîne réactive. La bibliothèque utilise Handler en interne et garantit l'exécution sur le thread UI même sous forte charge. Pour les opérations d'arrière-plan, Schedulers.io() supporte un pool de threads illimité et convient à toutes les opérations bloquantes. Schedulers.computation() utilise un pool fixe égal au nombre de cœurs CPU.

RxJava dans Android : applications pratiques

RxJava dans Android est utilisé pour trois scénarios principaux : les requêtes réactives à Room, l'intégration avec Retrofit et la liaison réactive UI via RxBinding. Chaque scénario a son propre ensemble de types : Room retourne Flowable pour les requêtes observables, Retrofit retourne Single pour les requêtes HTTP, RxBinding retourne Observable pour les événements UI.

Room + RxJava

Room est une bibliothèque de persistance de données de Google. À partir de Room 2.1, la base de données supporte les types de retour réactifs : Flowable et Observable. Lorsqu'un enregistrement dans la table change, Room envoie automatiquement une nouvelle valeur dans le flux. Le développeur s'abonne à Flowable dans le ViewModel et reçoit des données à jour sans requêtes manuelles à chaque changement.

// Room DAO avec requête réactive
@Dao
interface UserDao {
    @Query("SELECT * FROM users WHERE id = :id")
    fun getUserById(@Param("id") userId: Int): Flowable<User>

    @Insert
    fun insertUser(user: User): Completable
}

// ViewModel — composition Room + Network
class UserViewModel(private val dao: UserDao) : ViewModel() {
    val users: Flowable<List<User>> = dao.getAllUsers()
        .subscribeOn(Schedulers.io())
}

Le pattern MVVM + RxJava est construit sur le principe que le ViewModel n'a pas de références à la Vue. Le ViewModel publie des sources réactives (Flowable, LiveData via Transformations), et l'Activity ou le Fragment s'y abonne. Cela offre une testabilité : le ViewModel est testé sans l'UI en remplaçant les Schedulers via RxJavaPlugins.setComputationScheduler. CompositeDisposable dans le ViewModel gère le cycle de vie des abonnements — à onCleared(), tous les abonnements sont annulés.

RxJava vs Kotlin Flow

Kotlin Flow est une implémentation native de flux froids en Kotlin, intégrée aux coroutines et introduite dans Kotlin 1.3. Flow résout les mêmes problèmes que RxJava mais avec des différences fondamentales : support intégré des coroutines (fonctions suspend), annulation via coroutine cancellation et absence de problèmes de contre-pression — Flow utilise suspend au lieu du buffering. Flow fait partie de la bibliothèque standard Kotlin, ne nécessitant aucune dépendance supplémentaire.

RxJava reste le choix préféré pour les projets Java, les projets supportant Java 7-8 et les bases de code existantes en RxJava. L'écosystème RxJava est significativement plus riche : plus de 400 opérateurs contre environ 50 dans Flow, intégration avec Retrofit via un CallAdapter intégré, support de la contre-pression via Flowable, et RxBinding, RxPermissions, RxLocation pour Android. Kotlin Flow rattrape rapidement son retard, mais la flexibilité de RxJava dans les scénarios complexes de combinaison de flux est encore supérieure.