RxJava: basi, ReactiveX e lavoro con flussi di dati

Cos'è RxJava e ReactiveX

RxJava è l'implementazione Java della specifica ReactiveX, una libreria per la programmazione asincrona che utilizza flussi osservabili (Observable). RxJava 2 è stato rilasciato nel 2016 con il supporto di Reactive Streams (Flowable) e la separazione in rx.Observable e io.reactivex.Observable. RxJava 3 (2019) è la versione principale attuale con compatibilità all'indietro con RxJava 2.

L'idea centrale di RxJava è che tutto è un flusso: flusso di dati, flusso di eventi, flusso di stati. Qualsiasi operazione asincrona può essere rappresentata come un Observable che emette dati, un errore o un segnale di completamento. Un Observer si sottoscrive all'Observable e riceve notifiche in tempo reale.

Secondo Badoo (2024), prima della transizione alle coroutine, il 76% delle app Android nella top 200 di Google Play utilizzava RxJava per operazioni asincrone. La quota sta diminuendo a favore delle coroutine, ma RxJava rimane nel codice di produzione di migliaia di app ed è considerata una tecnologia matura e collaudata. ReactiveX è una specifica multipiattaforma implementata anche per JavaScript (RxJS), .NET (Rx.NET), Swift (RxSwift) e altri linguaggi.

Pattern Observer in RxJava

ReactiveX estende il pattern Observer classico con due meccanismi: concatenamento di operatori e threading basato su Schedulers. Observable non inizia a emettere dati finché un Observer non si sottoscrive (valutazione pigra). Ciò consente di costruire una pipeline di dati che si attiva solo quando esiste una sottoscrizione.

Tipi di Observable: Observable, Flowable, Single, Maybe, Completable

Observable — il tipo base che emette 0..N elementi con onError o onComplete. Adatto per flussi di dati illimitati — ad esempio, eventi di clic o aggiornamenti di geolocalizzazione. Observable non supporta il backpressure.

Flowable — la versione Reactive Streams di Observable con supporto backpressure. Utilizzato quando la fonte di dati può generare elementi più velocemente di quanto l'Observer possa elaborarli. Flowable supporta le strategie BACKPRESSURE_BUFFER, DROP, LATEST e ERROR.

Single, Maybe e Completable

Single emette esattamente un elemento o un errore — ideale per richieste di rete. Maybe emette 0 o 1 elemento, adatto per cache dove i dati potrebbero essere assenti. Completable emette solo onComplete o onError, senza dati, comodo per operazioni di scrittura o eliminazione. Questi tipi semplificano l'API restringendo il contratto a un caso specifico. Retrofit (un popolare client HTTP per Android) supporta direttamente tutti e cinque i tipi RxJava, consentendo di scegliere il tipo di ritorno più appropriato per ogni endpoint senza codice boilerplate aggiuntivo.

Operatori RxJava: trasformazione e filtraggio dei flussi

Gli operatori sono funzioni che trasformano un Observable in un altro. Una catena di operatori descrive la pipeline di dati: ogni operatore prende il flusso dal precedente, lo trasforma e lo passa al successivo. RxJava contiene oltre 200 operatori raggruppati in categorie.

• map — trasforma ogni elemento (Integer → String)
• flatMap — trasforma un elemento in un Observable e fonde tutto in un unico flusso
• filter — fa passare gli elementi che soddisfano una condizione
• zip — combina elementi di N Observable per indice
• merge — fonde più Observable in uno, preservando l'ordine cronologico
• debounce — emette elementi solo se è trascorso un intervallo di tempo specificato senza altra emissione

flatMap è uno degli operatori più potenti di RxJava. Consente di eseguire una richiesta asincrona per ogni elemento e raccogliere i risultati in un flusso comune. Ad esempio, flatMap viene utilizzato per caricare i dettagli da un elenco di ID: ogni ID → richiesta di rete → fusione dei risultati. A differenza di map, che semplicemente trasforma un elemento, flatMap può emettere più elementi o passare a un altro Observable, rendendolo la base per la costruzione di pipeline asincrone.

Gestione degli errori con operatori

onErrorResumeNext — passa a un Observable di backup in caso di errore. retry — si risottoscrive N volte in caso di errore. onErrorReturn — restituisce un valore predefinito invece dell'errore. doOnError — esegue un effetto collaterale in caso di errore senza modificare il flusso (registrazione o analisi). La combinazione di questi operatori consente di costruire pipeline robuste con una chiara strategia di gestione degli errori senza try/catch manuale.

Schedulers: gestione dei thread in RxJava

Gli Schedulers determinano su quale thread vengono eseguiti Observable e Observer. subscribeOn imposta il thread per la fonte, observeOn imposta il thread per l'Observer e gli operatori successivi. Questa separazione è un vantaggio chiave di RxJava: fonte sul thread IO, elaborazione su computation, UI sul thread principale.

Principali Schedulers: Schedulers.io() — per operazioni I/O (rete, disco), pool illimitato. Schedulers.computation() — per calcoli, pool fisso in base al numero di core. Schedulers.newThread() — un nuovo thread per ogni attività. AndroidSchedulers.mainThread() — thread principale Android (RxAndroid). Esiste anche Schedulers.trampoline() per eseguire attività nel thread corrente con una coda FIFO, utile per i test.

Secondo Google (2025), l'uso corretto degli Schedulers è la parte più difficile di RxJava per i principianti. Un errore tipico è chiamare subscribeOn dopo observeOn, che non influisce sulla fonte. subscribeOn dovrebbe essere il primo nella catena per la fonte, observeOn prima della sottoscrizione UI. Regola: subscribeOn influisce solo sull'upstream (fonte), observeOn cambia il downstream (sottoscrittore e tutti gli operatori successivi).

Esempi di codice RxJava in Android

Consideriamo tre scenari: una richiesta di rete con Single, richieste parallele con zip e debounce per un campo di ricerca con debounce.

Richiesta di rete con Single

Single è perfetto per le richieste Retrofit: una richiesta — una risposta. Sottoscrivi sul thread principale per gli aggiornamenti UI.

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); }
    })

Richieste parallele con zip

zip combina i risultati di due Single indipendenti in uno. Vengono eseguiti in parallelo, il risultato viene prodotto dopo il completamento di entrambi.

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 per campo di ricerca

debounce ignora le modifiche rapide del testo e invia una richiesta solo dopo una pausa di 400 ms. distinctUntilChanged annulla la richiesta se il testo non è cambiato.

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: confronto degli approcci

RxJava e Kotlin Coroutines risolvono lo stesso problema — la programmazione asincrona — ma con approcci fondamentalmente diversi. RxJava è costruito sul pattern Observer ed è basato sul push: la fonte invia dati, l'Observer reagisce. Le coroutine sono basate sul pull: il codice richiede dati in sequenza tramite await.

• RxJava — reattivo, flusso di dati, >200 operatori, push-based, curva di apprendimento ripida
• Coroutines — sequenziale, suspend/await, ~40 funzioni, pull-based, sintassi semplice
• RxJava — maturo (2016), ecosistema enorme, ma curva di apprendimento ripida
• Coroutines — moderno (2018), scelta preferita di Google per il nuovo codice
• RxJava — backpressure integrato tramite Flowable, strategie di buffer ben testate
• Coroutines — Flow con backpressure è recente, ma attivamente sviluppato da JetBrains

Secondo Google I/O 2024, Kotlin Coroutines è l'approccio consigliato per il nuovo codice asincrono in Android. RxJava rimane supportato per i progetti esistenti. Google fornisce librerie ponte (kotlinx-coroutines-rx3) per una migrazione graduale. AndroidX (LiveData, Room, Paging 3) supporta entrambi gli approcci, consentendo di utilizzare RxJava nei moduli vecchi e le coroutine in quelli nuovi senza conflitti di dipendenze.

Strategia di migrazione da RxJava alle coroutine

Transizione graduale: ogni nuovo componente viene scritto con coroutine, il vecchio codice RxJava non viene modificato. RxJava → coroutine tramite awaitSingle() o awaitFirst(). Coroutine → RxJava tramite future() o asFlowable(). La migrazione completa richiede 6–18 mesi per progetti grandi.