RxJava: essenza, componenti e programmazione reattiva

Cos'è RxJava?

RxJava è un'implementazione della libreria ReactiveX (Reactive Extensions) per la Macchina Virtuale Java. La prima versione di RxJava è stata rilasciata da Netflix nel 2013 per gestire chiamate asincrone in applicazioni lato server. Al momento della sua creazione, le principali alternative in Java erano Future e Callback — entrambi gli approcci portavano a callback-hell e a una gestione complessa dei thread. RxJava ha introdotto la composizione di operazioni asincrone tramite Observable con catene di operatori funzionali.

L'architettura di RxJava si basa sulla specifica Reactive Streams — uno standard per l'elaborazione asincrona di flussi con contropressione non bloccante. La specifica definisce quattro interfacce: Publisher, Subscriber, Subscription e Processor. RxJava 2+ implementa completamente Reactive Streams attraverso il tipo Flowable, rispettando i contratti di contropressione a differenza di RxJava 1. Observable in RxJava 2 non supporta la contropressione — è destinato a flussi con un numero ridotto di eventi o eventi UI.

Secondo il sondaggio JetBrains, 2025, RxJava è tra le 3 librerie principali per lo sviluppo Android. I casi d'uso principali includono: gestione delle richieste di rete tramite Retrofit (integrato con RxJava via CallAdapter), lavoro con Room (le query reattive restituiscono Flowable o Maybe), animazioni ed eventi UI tramite RxBinding e ricerca con debounce sull'input di testo. Tutti questi scenari condividono un modello di catena comune: sorgente (Observable) → trasformazione (operatori) → sottoscrizione (subscribe).

Storia delle versioni di RxJava

RxJava 1 (2013) ha gettato le basi con Observable e operatori, ma soffriva di problemi di contropressione — nei flussi veloci, i dati si accumulavano in memoria, causando OutOfMemoryError. RxJava 2 (2016) ha corretto l'architettura separando Observable (senza contropressione) e Flowable (con contropressione). RxJava 3 (2020) ha aggiunto il supporto per Java 8 Stream API, operatori aggiuntivi e prestazioni di sottoscrizione migliorate. Attualmente, RxJava 3 è la versione consigliata per i nuovi progetti.

Tipi di flussi reattivi in RxJava

RxJava fornisce cinque tipi principali di sorgenti reattive, ciascuna progettata per uno scenario specifico. Observable e Flowable emettono più valori, Single emette un valore o un errore, Completable emette solo il completamento senza dati, e Maybe emette un valore, zero o un errore. Scegliere il tipo corretto riduce il volume di codice e rende la catena auto-documentante.

Flowable è il tipo più flessibile per lavorare con grandi flussi di dati. Implementa il Publisher di Reactive Streams con supporto della contropressione: il consumatore può richiedere un numero specifico di elementi tramite Subscription.request(n). Ciò previene l'overflow del buffer quando le velocità di produttore e consumatore non corrispondono. Se la contropressione non è critica, usa Observable — ha meno overhead a causa dell'assenza del meccanismo di request.

Single è la scelta ottimale per le richieste HTTP. Retrofit 2 con RxJava CallAdapter restituisce Single<ResponseBody> per ogni richiesta. Single garantisce esattamente una chiamata a onSuccess o onError, corrispondendo alla semantica di una richiesta HTTP — una risposta o un errore. Completable è usato per operazioni di scrittura che non restituiscono dati: insert, update, delete. Maybe è comodo per il controllo della cache — può restituire un valore o meno.

// Esempio di utilizzo di Single per una richiesta HTTP
interface ApiService {
    @GET("users/{id}")
    fun getUser(@Path("id") userId: Int): Single<User>
}

// Sottoscrizione con elaborazione sul thread principale
apiService.getUser(42)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ user ->
        textView.text = user.name
    }, { error ->
        Log.e("API", "Error: ${error.message}")
    })
    .addTo(compositeDisposable)

Operatori di trasformazione e gestione dei flussi

Gli operatori in RxJava sono funzioni di ordine superiore che prendono una sorgente reattiva e ne restituiscono un'altra, trasformando il flusso di dati. RxJava 3 contiene oltre 400 operatori, divisi in categorie: trasformazione, filtraggio, combinazione, gestione degli errori e gestione del tempo. Ogni operatore è pigro — la catena viene costruita alla dichiarazione ed eseguita alla sottoscrizione.

Operatori di trasformazione

map è l'operatore di base che trasforma ogni valore tramite una funzione. flatMap prende una funzione che restituisce un Observable per ogni elemento e appiattisce il risultato in un unico flusso. switchMap è simile a flatMap, ma quando arriva un nuovo elemento, si disiscrive dall'Observable precedente. concatMap preserva l'ordine degli elementi — a differenza di flatMap, si sottoscrive sequenzialmente a ogni Observable annidato.

// Analisi JSON con trasformazione e filtraggio
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("Errore", it.message) })

Combinazione di flussi è un'area in cui RxJava eccelle particolarmente. zip combina elementi da più Observable a coppie per indice: primo con primo, secondo con secondo. combineLatest emette un nuovo valore quando qualsiasi flusso cambia, combinando i valori più recenti di tutti i flussi. merge combina più Observable in uno, preservando l'ordine di arrivo degli eventi. concat si sottoscrive sequenzialmente a ogni Observable e passa tutti i suoi eventi prima di passare al successivo.

Gestione del tempo include debounce (attendere una pausa nel flusso prima di emettere), throttleFirst (emettere il primo evento, ignorare i restanti entro una finestra), timeout (errore se nessun evento arriva nell'intervallo). La ricerca con debounce sull'input di testo è lo scenario più comune: searchObservable.debounce(300, MILLISECONDS).distinctUntilChanged() previene richieste non necessarie durante la digitazione veloce.

Scheduler e multithreading

Scheduler in RxJava è un'astrazione su un pool di thread. La libreria fornisce cinque Scheduler integrati: Schedulers.io() per operazioni di I/O (rete, file), Schedulers.computation() per attività intensive di CPU, Schedulers.newThread() per un nuovo thread ogni volta, Schedulers.single() per esecuzione a thread singolo e Schedulers.trampoline() per esecuzione immediata nel thread corrente.

subscribeOn e observeOn

subscribeOn determina su quale Scheduler viene eseguito l'Observable sorgente. Se ci sono più subscribeOn nella catena, la priorità va al più vicino alla sorgente. observeOn cambia il downstream allo Scheduler specificato — ogni uso di observeOn cambia il thread per gli operatori successivi. Un pattern Android tipico: subscribeOn(Schedulers.io()) per operazioni di rete, observeOn(AndroidSchedulers.mainThread()) per aggiornamenti UI.

// Elaborazione multithread con cambio di contesto
Observable.fromCallable(() -> database.getItems())
    .subscribeOn(Schedulers.io())            // DB su io
    .map(items -> processItems(items))     // trasformazione su io
    .observeOn(Schedulers.computation())    // passare a computation
    .map(processed -> compressImages(processed))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> ui.showResult(result))

AndroidSchedulers.mainThread() è uno Scheduler della libreria RxAndroid che esegue codice sul thread principale di Android. È obbligatorio per qualsiasi aggiornamento UI in una catena reattiva. La libreria usa Handler internamente e garantisce l'esecuzione sul thread UI anche sotto carico elevato. Per operazioni in background, Schedulers.io() supporta un pool di thread illimitato ed è adatto a qualsiasi operazione bloccante. Schedulers.computation() usa un pool fisso pari al numero di core CPU.

RxJava in Android: applicazioni pratiche

RxJava in Android è utilizzato per tre scenari principali: query reattive a Room, integrazione con Retrofit e binding reattivo UI tramite RxBinding. Ogni scenario ha il proprio insieme di tipi: Room restituisce Flowable per query osservabili, Retrofit restituisce Single per richieste HTTP, RxBinding restituisce Observable per eventi UI.

Room + RxJava

Room è una libreria di persistenza dati di Google. A partire da Room 2.1, il database supporta tipi di ritorno reattivi: Flowable e Observable. Quando un record nella tabella cambia, Room invia automaticamente un nuovo valore nel flusso. Lo sviluppatore si sottoscrive a Flowable nel ViewModel e riceve dati aggiornati senza query manuali a ogni modifica.

// Room DAO con query reattiva
@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 — composizione Room + Network
class UserViewModel(private val dao: UserDao) : ViewModel() {
    val users: Flowable<List<User>> = dao.getAllUsers()
        .subscribeOn(Schedulers.io())
}

Il pattern MVVM + RxJava si basa sul fatto che il ViewModel non ha riferimenti alla View. Il ViewModel pubblica sorgenti reattive (Flowable, LiveData tramite Transformations), e l'Activity o il Fragment vi si sottoscrivono. Ciò offre testabilità: il ViewModel viene testato senza UI, sostituendo gli Scheduler tramite RxJavaPlugins.setComputationScheduler. CompositeDisposable nel ViewModel gestisce il ciclo di vita delle sottoscrizioni — a onCleared(), tutte le sottoscrizioni vengono cancellate.

RxJava vs Kotlin Flow

Kotlin Flow è un'implementazione nativa di flussi freddi in Kotlin, integrata nelle coroutine e introdotta in Kotlin 1.3. Flow risolve gli stessi problemi di RxJava ma con differenze fondamentali: supporto integrato delle coroutine (funzioni suspend), cancellazione tramite coroutine cancellation e assenza di problemi di contropressione — Flow usa suspend invece del buffering. Flow fa parte della libreria standard Kotlin, senza richiedere dipendenze aggiuntive.

RxJava rimane la scelta preferita per progetti Java, progetti con supporto Java 7-8 e basi di codice esistenti in RxJava. L'ecosistema RxJava è significativamente più ricco: oltre 400 operatori contro circa 50 in Flow, integrazione con Retrofit tramite CallAdapter integrato, supporto della contropressione tramite Flowable e RxBinding, RxPermissions, RxLocation per Android. Kotlin Flow sta recuperando rapidamente, ma la flessibilità di RxJava in scenari complessi di combinazione di flussi è ancora superiore.