Questo capitolo completa la seconda parte del corso, dedicata alla programmazione asincrona: dopo il multi-threading sincrono della prima parte (componenti attivi con sequenze di azioni potenzialmente bloccanti, sospensione del thread gestita dal sistema operativo), il percorso asincrono ed event-driven — event loop, callback, Promise, async/await e coroutine — approda ora al paradigma reattivo con le Reactive Extensions. I componenti non si bloccano mai in attesa: reagiscono a stimoli esterni — richieste di rete, input da sensori, azioni dell'utente — e ogni operazione asincrona spezza il flusso in avvio e continuazione.
Con le Promise assumiamo che le computazioni asincrone producano un singolo risultato. Ma nelle applicazioni reali è frequente dover gestire flussi di dati/eventi asincroni: click del mouse, letture da un sensore, messaggi in arrivo da una connessione di rete. La programmazione reattiva (RP) nasce per questo: è la programmazione con flussi di dati asincroni.
La programmazione reattiva è un paradigma orientato ai flussi di dati e alla propagazione del cambiamento. Le variabili non hanno più un valore fisso: diventano reattive, aggiornandosi automaticamente quando cambiano le variabili da cui dipendono. È come se l'assegnamento non fosse più un'operazione imperativa ma la dichiarazione di una relazione tra quantità.
Le radici della RP risalgono al linguaggio Fran (Elliott & Hudak, 1997), un dominio-specifico funzionale per animazioni interattive. Da lì nasce la Functional Reactive Programming (FRP), basata sul synchronous dataflow (Lee & Messerschmitt, 1987) ma con vincoli real-time rilassati. Oggi la RP è implementata da framework e librerie per tutti i linguaggi mainstream: FrTime (Scheme), Reactive Extensions di Microsoft (.NET, RxJS, RxJava), Flapjax, Bacon.js, Scala.React, e molti altri.
Il paper di riferimento citato dal professore è il survey di Bainomugisha et al. (2013) su ACM Computing Surveys. Altri lavori importanti includono "Your mouse is a database" di Erik Meijer (2012) e i lavori di Salvaneschi sulla distributed RP (OOPSLA 2014).
Singolo risultato. La computazione asincrona produce un valore (o un errore) una volta sola. La composizione avviene tramite chaining di then o callback annidate. Il flusso è "freddo" nel senso che ogni then crea una nuova Promise.
Singolo risultato ma sintassi sincrona. L'operatore await sospende la funzione finché la Promise non è risolta, permettendo di scrivere codice sequenziale lineare anche se asincrono. Il blocco non è più atomico.
Flussi di valori nel tempo. Non un risultato singolo, ma una sequenza potenzialmente infinita di eventi (stream). Ogni evento può essere trasformato, filtrato, combinato con altri flussi usando operatori dichiarativi (map, filter, merge, zip). La propagazione del cambiamento è automatica.
L'idea centrale della RP è la propagazione automatica del cambiamento. Consideriamo l'esempio più semplice: se dichiaro var3 = var1 + var2, in un linguaggio reattivo var3 viene aggiornato automaticamente ogni volta che cambia var1 o var2.
flowchart LR
var1["var1 = 1"] --> var3["var3 = var1 + var2 = 3"]
var2["var2 = 2"] --> var3
Grafo delle dipendenze: var3 dipende da var1 e var2. Ogni volta che var1 o var2 cambiano, var3 viene automaticamente ricalcolato. Il framework costruisce e gestisce questo grafo dietro le quinte.
Il professore usa una metafora efficace: non stiamo più scrivendo un assegnamento (imperativo), ma stiamo dichiarando una relazione matematica tra quantità. Come dice il professore: "L'uguale in matematica non è un assegnamento, esprime una relazione che permette di affermare sempre il fatto che due quantità sono uguali."
Questa è la differenza fondamentale tra programmazione imperativa e reattiva. Nella programmazione imperativa scrivete come calcolare un valore; nella programmazione reattiva dichiarate cosa deve essere quel valore in ogni istante. Il "quando" lo gestisce automaticamente il framework.
Il professore si chiede: "Col modello che conosciamo finora, come implementeremmo questa propagazione?" La risposta è: con gli eventi. Ogni variabile è sorgente di eventi, e ogni volta che cambia, notifica i dipendenti che si aggiornano. Ma nella RP questo meccanismo è automatico e dichiarativo: non gestite eventi, definite relazioni. Il framework si occupa della notifica, della schedulazione degli aggiornamenti e della consistenza.
Il professore mette in guardia: se si rendesse osservabile uno stato intermedio in cui var1 è stato aggiornato ma var2 no, il sistema si troverebbe in una situazione inconsistente. Serve quindi che lo stato globale sia sempre consistente con le relazioni dichiarate. Questo è il problema dei glitch, che vedremo tra poco.
La RP si basa su due astrazioni fondamentali:
Valori continui nel tempo. Rappresentano un flusso ininterrotto di dati che ha sempre un valore definito in ogni istante. Un esempio è la temperatura di un sensore: è sempre definita, cambia nel tempo in modo continuo. Un timer che produce un tick ogni 100ms è un behaviour.
Valori discreti (o sparsi) nel tempo. Rappresentano eventi che occorrono in istanti specifici. Un esempio è il click del mouse: non è sempre definito, occorre quando l'utente clicca. Una sequenza di rilevazioni da un sensore di prossimità su un nastro trasportatore è un event stream.
Entrambe le astrazioni sono first-class values: possono essere passate a funzioni, combinate, trasformate. Il framework o linguaggio gestisce automaticamente la propagazione del cambiamento. Come dice il professore: "astrazioni che andiamo a manipolare e combinare sono flussi asincroni di dati che possono rappresentare eventi, valori che variano nel tempo, anche continui."
// Crea un behaviour timer che produce un valore ogni 100ms
var timer = timerB(100);
// Crea un secondo behaviour che arrotonda al secondo
var seconds = liftB(
function (time){
return Math.floor(time / 1000);
}, timer
);
// Inserisce il behaviour in un elemento DOM: si aggiorna automaticamente!
insertDomB(seconds, 'timer-div');Questo esempio è completamente dichiarativo. Non ci sono callback, non ci sono event listener: definite la relazione seconds = floor(timer / 1000) e il framework si occupa di aggiornare il DOM automaticamente ogni volta che il timer scatta.
Nella programmazione imperativa dovreste: (1) creare un interval timer, (2) registrare un callback, (3) calcolare secondi, (4) aggiornare il DOM. Nella RP dichiarate semplicemente la relazione: il DOM contiene floor(timer / 1000) e tutto si aggiorna da sé. È la stessa differenza tra dire "sposta il tappeto" (imperativo) e dire "il tappeto deve essere sempre sotto il tavolo" (dichiarativo/reattivo).
Come avviene la propagazione del cambiamento? Il professore distingue due modelli fondamentali:
Il produttore, quando ha nuovi dati, li spinge (push) ai consumatori dipendenti. La propagazione è guidata dalla disponibilità di nuovi dati. Tipico dei linguaggi eager (JavaScript, Scala). Più efficiente per flussi con aggiornamenti frequenti, ma può portare a valutazioni ridondanti se il consumatore non è pronto. Esempi: Flapjax (basato su JavaScript), Scala.React.
Il consumatore, quando necessita di un valore, lo tira (pull) dal produttore. La propagazione è guidata dalla domanda (demand-driven). Tipico dei linguaggi lazy (Haskell, tramite lazy evaluation). Più efficiente quando il consumatore decide quando aggiornarsi e non vuole essere sommerso da notifiche non richieste.
I glitch sono inconsistenze temporanee che possono verificarsi durante la propagazione del cambiamento, specialmente nel modello push. Si verificano quando una computazione viene eseguita prima che tutte le sue dipendenze siano state aggiornate.
flowchart TD
subgraph "t1: stato consistente"
v1["var1 = 1"] --> v2["var2 = var1 * 1 = 1"]
v1 --> v3["var3 = var1 + var2 = 2"]
v2 --> v3
end
flowchart TD
subgraph "t2: GLITCH! var1=2 ma var2 non ancora"
v1["var1 = 2"] -.-> v2["var2 = 1 (NON aggiornato!)"]
v1 -.-> v3["var3 = 2 + 1 = 3 (SBAGLIATO!)"]
v2 -.-> v3
style v3 fill:#fef2f2,stroke:#ef4444
end
flowchart TD
subgraph "t3: stato consistente"
v1["var1 = 2"] --> v2["var2 = 2"]
v1 --> v3["var3 = 2 + 2 = 4"]
v2 --> v3
end
La sequenza mostra il problema: quando var1 cambia da 1 a 2:
var1 viene aggiornato a 2. La propagazione potrebbe raggiungere var3 (via var1 + var2) prima che var2 venga ricalcolato. Risultato: var3 = 2 + 1 = 3. È un glitch: uno stato che non corrisponde a nessuna configurazione consistente delle relazioni.var2 viene aggiornato a 2, var3 viene ricalcolato a 4. Ora lo stato è consistente.I framework reattivi moderni implementano tecniche di glitch avoidance (tipicamente basate su topological ordering o livelli di propagazione) per garantire che uno stato inconsistente non sia mai osservabile dall'esterno. La glitch avoidance in contesti distribuiti è ancora un problema di ricerca aperto, a causa di latenze di rete, guasti e assenza di un clock globale, come discusso nei lavori di Salvaneschi et al. (2014) e Mogk et al. (2018).
Il lifting è il processo che converte una variabile o funzione normale in una reattiva. Quando una variabile viene assegnata con un'espressione che coinvolge behaviour o event stream, diventa a sua volta reattiva: qualsiasi cambiamento nei flussi da cui dipende si propaga automaticamente.
// timer behaviour
var timer = timerB(100);
// seconds è ottenuto "sollevando" (lifting) una funzione normale su timer
// La funzione Math.floor(time/1000) viene "liftata" a operare su behaviour
var seconds = liftB(function (time) {
return Math.floor(time / 1000);
}, timer);Il framework costruisce automaticamente un grafo delle dipendenze tra le variabili reattive. Quando un behaviour cambia, tutte le variabili che dipendono da esso vengono ricalcolate. Se una variabile A dipende da B, e B dipende da C, quando C cambia, A viene ricalcolata solo dopo che B è stato aggiornato (se il framework implementa glitch avoidance).
Alcuni framework fanno lifting implicitamente (Bacon.js), altri richiedono che il programmatore lo esegua manualmente (React.js), altri offrono entrambe le modalità (Flapjax: come libreria vuole lifting esplicito, come compilatore trasforma il codice implicitamente).
La capacità di comporre flussi è ciò che permette di evitare il callback hell. Invece di avere tre callback separate per gestire click, movimento e rilascio del mouse, possiamo comporle in un unico flusso:
// Composizione di event stream in Flapjax
var saveTimer = timerE(10000); // event stream: ogni 10 secondi
var saveClicked = extractEventE('save-button', 'click'); // event stream: click
var save = mergeE(saveTimer, saveClicked); // merge dei due flussi
save.mapE(doSave); // per ogni evento nel flusso composito, chiama doSaveIl professore illustra gli operatori principali per la manipolazione dei flussi, come mostrato nelle slide:
map: trasforma ogni elemento del flusso applicando una funzione. Dato un flusso di numeri, ne produce uno di stringhe. È l'operatore fondamentale per trasformare i dati.
// Flapjax: da un flusso di numeri a un flusso di stringhe
var seqNum = Observable.range(1, 5);
var seqString = seqNum.map(n => new String('*').repeat(n));
seqString.subscribe(str => { console.log(str); });filter: seleziona solo gli elementi che soddisfano un predicato booleano. Dato un flusso misto di cerchi e quadrati, produce un nuovo flusso con i soli cerchi.
// Flapjax: filtra solo i cerchi
var cerchi = forme.filter(f => f.tipo === 'cerchio');flatMap: per ogni elemento genera un nuovo flusso, e tutti i flussi generati vengono appiattiti (flatten) in un unico flusso. È una combinazione di map e merge. Se per ogni palla generate due rombi, flatMap vi darà un unico flusso con tutti i rombi, indipendentemente dal colore della palla originale. Molto utile nelle applicazioni reali.
merge: combina due o più flussi in uno solo, alternando gli elementi nell'ordine in cui arrivano. Nel tempo, gli elementi dei due flussi vengono mescolati. Il flusso risultante completa quando tutti i flussi sorgente hanno completato.
zip: accoppia gli elementi di due flussi a due a due, in modo sincrono. Prende il primo elemento del primo flusso e lo accoppia con il primo elemento del secondo flusso, poi il secondo col secondo, e così via. È un operatore di sincronizzazione tra flussi.
Il professore commenta ironicamente che "alcuni programmi sono diventati illeggibili perché tutto è espresso come composizione di operatori" — ma questo è un problema di abuso, non del paradigma in sé.
Il punto di partenza è lo stesso visto nell'introduzione: sia le CPS (Continuation Passing Style) sia le Promise gestiscono computazioni asincrone che producono un singolo risultato, ma nelle applicazioni moderne è sempre più comune dover gestire flussi di dati asincroni (streams). La programmazione reattiva è fortemente legata al pattern Observer e alla programmazione event-driven: introdotta negli anni '90, oggi è fondamentale per lo sviluppo di applicazioni web reattive e Big Data.
Il professore specifica: "Nothing happens until you subscribe." La configurazione della pipeline è dichiarativa; i dati iniziano a fluire solo quando un subscriber si collega.
Le Reactive Extensions (Rx) sono la concretizzazione più diffusa del paradigma reattivo. Nate su .NET negli anni 2000 come evoluzione del pattern Observer combinate con LINQ, oggi sono disponibili per tutti i linguaggi mainstream: RxJS, RxJava, RxScala, RxPython, e adottate da framework come Angular (sia frontend che backend).
Rx combina le idee migliori del pattern Observer (notifica di cambiamenti), del pattern Iterator (scorrimento di sequenze) e della programmazione funzionale (operatori filter, map, reduce sui flussi). Come dice il famoso articolo di Erik Meijer (2012): "Your mouse is a database" — eventi asincroni diventano dati che possono essere interrogati, filtrati e combinati con operatori LINQ, come faremmo con un database. Un mouse non è più solo un dispositivo di input: è un database di click e movimenti.
Le tre proprietà fondamentali di Rx sono:
Al cuore di Rx ci sono due interfacce, elegantemente semplici: un Observable<T> rappresenta un flusso di dati a cui ci si sottoscrive con un Observer<T>.
interface IObservable<T> {
IDisposable Subscribe(IObserver<T> observer);
}
interface IObserver<T> {
void OnNext(T value); // chiamato quando arriva un nuovo elemento
void OnError(Exception error); // chiamato in caso di errore
void OnCompleted(); // chiamato quando il flusso termina
}Un Observable emette eventi a cui un Observer si sottoscrive. La sottoscrizione restituisce un IDisposable che permette di cancellare la sottoscrizione quando non serve più. Ogni chiamata di sottoscrizione può potenzialmente avvenire su un thread diverso da quello in cui la sequenza viene eseguita.
Esempio concreto in .NET:
IObservable<int> source = Observable.Range(1, 10);
IDisposable subscription = source.Subscribe(
x => Console.WriteLine("OnNext: {0}", x),
ex => Console.WriteLine("OnError: {0}", ex.Message),
() => Console.WriteLine("OnCompleted")
);
// ...
subscription.Dispose(); // cancella la sottoscrizioneIl professore nota che la chiamata a Subscribe è asincrona: il chiamante non viene bloccato fino al completamento della sequenza. Il thread che chiama Subscribe può essere diverso dal thread in cui la sequenza viene eseguita.
A chiusura della lezione introduttiva, il professore offre una prima panoramica dei temi che le sezioni seguenti approfondiranno in dettaglio: operatori di composizione, backpressure e la distinzione hot/cold.
I flussi Observable possono essere trasformati e combinati con operatori ispirati a LINQ. Il professore mostra gli operatori principali con esempi concreti:
Concatena due sequenze in ordine: la seconda inizia solo dopo che la prima ha completato.
var source1 = Observable.Range(1, 3);
var source2 = Observable.Range(1, 3);
source1.Concat(source2)
.Subscribe(Console.WriteLine);
// Output: 1,2,3,1,2,3Fonde due sequenze: gli elementi appaiono nell'ordine in cui vengono emessi. Le due sequenze sono attive contemporaneamente.
var source1 = Observable.Range(1, 3);
var source2 = Observable.Range(1, 3);
source1.Merge(source2)
.Subscribe(Console.WriteLine);
// Output: 1,1,2,2,3,3Trasforma ogni elemento del flusso applicando una funzione di proiezione.
var seqNum = Observable.Range(1, 5);
var seqString = from n in seqNum
select new string('*', (int)n);
seqString.Subscribe(str => {
Console.WriteLine(str);
});Accoppia gli elementi di due flussi a due a due, in ordine. È un operatore sincrono: aspetta che entrambi i flussi abbiano un elemento disponibile per produrre una coppia. Se un flusso è più veloce dell'altro, gli elementi in eccesso vengono bufferizzati.
Altri operatori importanti menzionati dal professore includono operatori di filtraggio (Where, Any), operatori temporali (Throttle, Debounce, Delay), e operatori per la gestione delle eccezioni (Catch, Retry, Finally).
La backpressure è la capacità del consumatore di segnalare al produttore che la velocità di emissione è troppo alta. Il professore usa l'analogia della catena di montaggio: se una postazione di lavoro lavora più lentamente della precedente, segnala a monte di limitare il flusso.
Un aspetto importante che il professore sottolinea: nulla accade fino a quando non ci si sottoscrive (subscribe). Prima della subscribe c'è solo la fase di configurazione in cui si descrive il flusso dichiarativamente, costruendo la pipeline di operatori. La subscribe attiva effettivamente il flusso, e lo fa tramite un segnale di request che si propaga all'indietro fino alla sorgente. È la stessa distinzione configurazione/esecuzione che abbiamo visto con le Promise.
Rx distingue due tipi di flussi reattivi:
Ogni subscriber ottiene un nuovo flusso indipendente, partendo dall'inizio. Come un DVD: ogni spettatore che lo avvia parte dal primo minuto. Tutti i dati vengono generati per ogni subscriber separatamente.
Il flusso esiste indipendentemente dai subscriber. I subscriber in ritardo ricevono solo gli elementi emessi dopo la sottoscrizione. È come una diretta TV: se vi sintonizzate a metà, vedete solo da quel momento in poi.
Alcuni flussi hot possono cache o replay la storia delle emissioni (totalmente o parzialmente). Un flusso hot può emettere eventi anche quando nessun subscriber è in ascolto — un'eccezione alla regola "nothing happens before you subscribe". Questa distinzione è fondamentale per scegliere il tipo di flusso giusto in base allo scenario applicativo.
Il professore accenna anche alle direzioni di ricerca attuali nella programmazione reattiva:
I concetti chiave da portare all'esame: differenza tra hot e cold stream, backpressure e l'analogia della catena di montaggio, operatori principali (map, filter, merge, zip, concat, flatMap), la distinzione tra fase di configurazione e fase di esecuzione in Rx, e la differenza fondamentale tra programmazione reattiva (flussi di valori) e programmazione asincrona tradizionale (valori singoli).
Il professore chiude ricordando che i big data e le piattaforme di stream processing usano estensivamente questi concetti: specificare computazioni elaborate su più nodi in modo dichiarativo, dove la gestione della consistenza è ancora più complessa che in contesto single-node.
Il professor Ricci dedica gran parte della lezione successiva a esplorare il ricco ecosistema degli operatori messi a disposizione da ReactiveX (Rx). Gli operatori sono il cuore della programmazione reattiva: permettono di manipolare flussi di dati asincroni con lo stesso tipo di espressivita che si ha con le operazioni funzionali su collezioni (map, filter, reduce), ma con il valore aggiunto della gestione del tempo e dell'asincronia.
Gli operatori si classificano in diverse categorie, come presentato nelle slide e nei lab notes:
Il professor Ricci sottolinea un punto affascinante: l'insieme degli operatori di ReactiveX e Turing-equivalente. Cio significa che qualsiasi computazione puo essere espressa esclusivamente combinando flussi e operatori. Si puo persino implementare un costrutto if usando gli operatori giusti. Questo mostra la potenza espressiva del modello a flussi di dati.
Gli operatori di trasformazione sono quelli che modificano gli elementi emessi da un flusso. Il professore ne mostra diversi esempi concreti nei test predisposti per il corso.
L'operatore map applica una funzione a ciascun elemento del flusso e riemette il risultato. E l'equivalente reattivo di una trasformazione funzionale classica.
Flowable.range(1, 10)
.map(v -> v * v) // quadrato di ogni numero
.subscribe(v -> System.out.println("Quadrato: " + v));
// Output: 1, 4, 9, 16, 25, 36, 49, 64, 81, 100L'operatore scan applica una funzione cumulativa a ogni elemento, emettendo ogni risultato intermedio. E simile a reduce ma restituisce tutti i valori di passaggio, non solo quello finale.
Flowable.range(1, 5)
.scan((acc, v) -> acc + v)
.subscribe(System.out::println);
// Output: 1, 3, 6, 10, 15buffer raccoglie periodicamente gli elementi in pacchetti (liste) e li emette come blocchi, riducendo il numero di emissioni. window fa qualcosa di simile ma emette finestre che sono a loro volta Observable, utili per operazioni che richiedono la manipolazione di sottosequenze. groupBy divide il flusso in sottoflussi organizzati per chiave.
Il professor Ricci consiglia di sperimentare con questi operatori nei test del laboratorio: i file Test01, Test02 e successivi contengono esempi funzionanti con Flowable.range, map, filter, scan e zip. Eseguirli e fondamentale per capire la differenza tra operatori che sembrano simili.
FlatMap e forse l'operatore piu potente e pervasivo della programmazione reattiva. Il professor Ricci lo spiega con cura, perche e centrale nell'assignment del corso.
Mentre map trasforma ogni elemento in un nuovo valore, flatMap trasforma ogni elemento in un nuovo flusso e poi appiattisce il risultato in un unico flusso. Da un flusso di elementi si ottiene un flusso di flussi che viene "appiattito" in un singolo flusso di output.
Flowable.range(1, 10)
.flatMap(v -> Flowable.just(v * v) // per ogni v, crea un nuovo flusso…
.subscribeOn(Schedulers.computation())) // …eseguito su un thread diverso
.subscribe(System.out::println);
// I quadrati vengono emessi fuori ordine, perche ogni
// sottoflusso viaggia su un thread diverso!La flatMap e l'operatore che permette di parallelizzare computazioni indipendenti. Quando combini flatMap con subscribeOn, ogni elemento del flusso originale viene processato su un thread diverso, e i risultati vengono poi riaggregati. Vedremo in dettaglio questo pattern nella sezione dedicata.
Il professore lo descrive cosi: "Se ogni elemento di un flusso e a sua volta un flusso, uso la flatMap per appiattire." E l'operatore giusto quando, analizzando qualcosa, capita che ogni elemento generi un sottoproblema che a sua volta produce un flusso di risultati.
Gli operatori di filtro selezionano quali elementi del flusso far passare, basandosi su condizioni logiche o temporali.
Come nelle lambda funzionali, filter trattiene solo gli elementi che soddisfano un predicato.
Flowable.range(1, 20)
.filter(v -> v % 3 == 0) // solo multipli di 3
.map(v -> v * v)
.subscribe(System.out::println);
// Output: 9, 36, 81, 144, 225, 324take(n) emette solo i primi n elementi, poi si completa. skip(n) salta i primi n elementi e lascia passare i successivi. takeLast(n) e skipLast(n) lavorano simmetricamente dalla fine del flusso.
debounce e un operatore temporale di grande utilita pratica. Se un elemento viene generato troppo vicino nel tempo rispetto a un elemento precedente (entro una finestra temporale specificata), viene scartato. E il meccanismo classico usato nei sistemi embedded e nelle UI per evitare "rimbalzi": ad esempio, in un campo di ricerca che deve chiamare un'API, non vogliamo inviare una richiesta a ogni singolo carattere digitato, ma solo dopo che l'utente ha smesso di scrivere per un certo intervallo.
sequenceDiagram
participant S as Sorgente eventi
participant D as debounce(200ms)
participant O as Output
S->>S: evento 1 (t=0ms)
S->>S: evento 2 (t=50ms) - scartato!
S->>S: evento 3 (t=180ms) - scartato!
S->>S: evento 4 (t=300ms)
D->>O: passa evento 4 (ultimo, dopo 200ms di silenzio)
Altri operatori temporali includono sample (prende l'ultimo elemento a intervalli regolari), throttleFirst (prende il primo elemento in ogni finestra temporale) e timeout (segnala errore se non arrivano elementi entro un tempo limite).
Spesso e necessario lavorare con piu flussi contemporaneamente. Gli operatori di combinazione permettono di fonderli in modo sincrono o asincrono.
merge combina piu Observable in uno solo, fondendo le emissioni nell'ordine in cui arrivano. Se due flussi emettono valori in parallelo, il risultato sara l'interleaving delle sequenze. L'output e 1,1,2,2,3,3 (interleaving) e si completa solo quando tutti i flussi sorgente hanno completato.
Observable.range(1, 3).mergeWith(Observable.range(1, 3))
.subscribe(System.out::println); // 1,1,2,2,3,3zip accoppia elementi in modo sincrono: prende il primo elemento del primo flusso e il primo elemento del secondo flusso, li combina tramite una funzione, e produce il primo elemento del flusso risultante. Poi il secondo, e cosi via. L'output e 1,2,3,4,5 (sincrono: attende che entrambi i flussi abbiano prodotto l'elemento i-esimo prima di procedere).
Il professore fa un esempio concreto: un flusso di parole ("the", "quick", "brown", "fox") viene zippato con un flusso di numeri (1, 2, 3, 4, 5), producendo coppie come "1. the", "2. quick", "3. brown", ecc.
combineLatest e diverso: quando uno qualsiasi dei flussi sorgente emette un nuovo valore, l'operatore prende l'ultimo valore di ciascun flusso e li combina. E ideale quando si ha un insieme di fonti che si aggiornano indipendentemente e si vuole sempre lo stato piu recente di tutte.
switch converte un Observable che emette Observable in un singolo Observable che emette solo gli elementi dell'ultimo Observable emesso. Se arriva un nuovo flusso, si "disiscrive" dal precedente e si aggancia al nuovo — utile per scenari di ricerca in tempo reale dove ogni nuova richiesta deve cancellare la precedente.
Il professore cita anche l'operatore join, che combina elementi di due Observable usando finestre temporali sovrapposte, e groupJoin per composizioni piu complesse.
Il professore mostra in dettaglio l'operatore zip e la sua versione zipWith, che e un esempio di composizione tra flussi di tipo diverso.
L'esempio concreto mostrato a lezione:
// Sorgente 1: parole
Flowable<String> src1 = Flowable.just("the", "quick", "brown", "fox");
// Sorgente 2: numeri
Flowable<Integer> src2 = Flowable.range(1, 5);
// Zip: accoppia in modo sincrono e combina
src1.zipWith(src2, (parola, numero) -> numero + ". " + parola)
.subscribe(System.out::println);
// Output: "1. the", "2. quick", "3. brown", "4. fox"Nell'operatore zip, i tipi cambiano: da due flussi di tipo diverso (String e Integer) si produce un nuovo flusso di un terzo tipo (String), determinato dalla funzione di combinazione. Questa flessibilita tipale e una delle caratteristiche che rende gli operatori di ReactiveX cosi potenti per il data flow.
Il professore nota che la zip attende che entrambi i flussi abbiano il prossimo elemento pronto prima di emettere un output. Se un flusso si completa prima, gli elementi rimanenti dell'altro flusso vengono scartati (come nell'esempio, dove "fox" viene comunque zippato con 4, ma il flusso src2 ha 5 elementi, mentre src1 ne ha 4: l'ultimo elemento di src2 non ha corrispondenza).
Altri operatori di composizione menzionati: and/then/when (che usano Pattern e Plan come intermediari), startWith (premette una sequenza di elementi), e join (che usa finestre temporali).
ReactiveX utilizza i marble diagrams (diagrammi a biglie) per rappresentare visivamente il comportamento degli operatori. Sono uno strumento potentissimo sia per capire che per progettare flussi reattivi.
Un marble diagram mostra: in alto, il flusso di input (una linea temporale con palline colorate che rappresentano gli elementi); al centro, l'operatore con i suoi parametri; in basso, il flusso di output risultante. Le linee verticali indicano il completamento, le X indicano errori, le barre di ritardo mostrano intervalli temporali.
flowchart LR
subgraph Input
direction LR
a1((1)) --> a2((2)) --> a3((3)) --> a4((4)) --> a5((5))
end
subgraph Operatore
m[map: v -> v * v]
end
subgraph Output
direction LR
b1((1)) --> b2((4)) --> b3((9)) --> b4((16)) --> b5((25))
end
a1 --> m --> b1
a2 --> m --> b2
a3 --> m --> b3
a4 --> m --> b4
a5 --> m --> b5
I marble diagrams sono cosi importanti che la documentazione ufficiale di ReactiveX (reactivex.io/documentation/operators.html) li usa come primaria forma di documentazione per ogni operatore. Il professore consiglia di consultarli ogni volta che si ha un dubbio sul comportamento di un operatore: "Un'immagine vale piu di mille parole, e nel caso della programmazione reattiva, un marble diagram vale piu di mille righe di codice."
Uno degli aspetti piu importanti della programmazione reattiva e il controllo su quale thread viene eseguita ciascuna parte della catena di elaborazione. Questo e il compito degli Schedulers.
Due operatori fondamentali controllano il threading:
subscribeOn(<Scheduler>) — specifica su quale thread deve essere eseguito l'Observable a monte (la sorgente e tutte le operazioni fino al primo observeOn).observeOn(<Scheduler>) — specifica su quale thread deve essere eseguito il Subscriber a valle (le operazioni successive).Flowable.fromCallable(() -> {
Thread.sleep(1000); // simulazione computazione costosa
return "Done";
})
.subscribeOn(Schedulers.io()) // esecuzione su thread I/O
.observeOn(Schedulers.single()) // osservazione su thread singolo (es. GUI)
.subscribe(System.out::println, Throwable::printStackTrace);Il professore spiega che questo pattern permette di spostare computazioni bloccanti su thread di background, lasciando il thread principale (o quello della GUI) libero per gestire l'interattivita. E l'equivalente reattivo dell'uso di ExecutorService in Java classico, ma molto piu dichiarativo e composizionale.
Il professor Ricci mette in guardia: se si usa subscribeOn piu volte nella stessa catena, solo il primo ha effetto. Per cambiare thread a meta catena, bisogna usare observeOn. Questo e un errore comune tra i principianti.
ReactiveX mette a disposizione diversi tipi di Scheduler, ciascuno progettato per un carico di lavoro specifico:
| Scheduler | Uso | Pool di thread |
|---|---|---|
Schedulers.computation() | Operazioni CPU-intensive (calcoli, trasformazioni) | Pool fisso: tanti thread quanti i core della CPU |
Schedulers.io() | Operazioni I/O-bound (file, rete, database) | Pool dinamico: cresce e decresce in base al carico |
Schedulers.single() | Esecuzione sequenziale su un unico thread (es. aggiornamento GUI) | Singolo thread, FIFO |
Schedulers.from(Executor) | Pool personalizzato | Wrap di un Executor esistente |
Il professore fa un esempio concreto: subscribeOn(Schedulers.io()) per caricare immagini e observeOn(Schedulers.computation()) per elaborarle (come ridimensionamento o filtri), per poi eventualmente tornare al thread UI per la visualizzazione.
myObservableServices.retrieveImage(url)
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.computation())
.subscribe(bitmap -> processImage(bitmap));Nota: Schedulers.from(executor) permette di usare pool esistenti (ad esempio un pool fisso di 10 thread, a differenza di computation() che usa tanti thread quanti i core e io() che e illimitato).
Una distinzione fondamentale nell'universo ReactiveX e quella tra Cold e Hot Observable. Il professore la spiega con chiarezza nelle slide e nei lab notes.
Un Cold Observable inizia a emettere elementi solo quando riceve almeno un subscriber. Ogni subscriber riceve l'intera sequenza di elementi a partire dall'inizio, indipendentemente dal momento in cui si e iscritto. E un modello di tipo pull (lazy).
Esempi tipici: Observable.just(), Observable.fromIterable(), Observable.range() — tutti quelli che generano dati su richiesta.
Observable<String> cold = Observable.just("A", "B", "C");
cold.subscribe(s -> System.out.println("Sub1: " + s)); // A, B, C
cold.subscribe(s -> System.out.println("Sub2: " + s)); // A, B, C
// Entrambi i subscriber ricevono TUTTI gli elementi!Un Hot Observable emette elementi continuamente, indipendentemente dalla presenza di subscriber. I subscriber tardivi ricevono solo gli elementi emessi dopo la loro iscrizione. E un modello di tipo push: la sorgente produce dati al proprio ritmo, e gli observer devono tenere il passo.
Esempi: eventi del mouse, tick di timer, stream di prezzi azionari, messaggi da un websocket. Un Hot Observable puo emettere anche quando non c'e nessun subscriber.
L'eccezione alla regola "nothing happens until you subscribe" — con gli Hot Observable, le cose succedono sempre.
| Caratteristica | Cold | Hot |
|---|---|---|
| Inizio emissione | Quando arriva un subscriber | Immediatamente (o da un evento esterno) |
| Elementi ricevuti | Tutti, dall'inizio | Solo quelli dopo l'iscrizione |
| Modello | Pull (lazy) | Push (eager) |
| Isolamento | Ogni subscriber ha la sua sequenza | Tutti i subscriber condividono la stessa sequenza |
| Backpressure | Gestibile a monte | Richiede strategie esplicite |
Alcuni Hot Observable possono cache o replay la storia delle emissioni, totalmente o parzialmente, permettendo ai subscriber tardivi di recuperare elementi passati.
Nella lezione di ripresa, il professore rivede i due concetti con i diagrammi di sequenza: la distinzione tra stream hot e cold, e il meccanismo di backpressure (che le prossime sezioni approfondiscono).
Un flusso cold inizia a generare elementi solo quando viene osservato (subscribed). Ogni subscriber riceve tutti gli elementi del flusso dal principio, indipendentemente dal momento in cui si sottoscrive. E come un file: se lo leggi da capo, vedi tutti i dati.
Comportamento: ogni subscriber ottiene una nuova sequenza indipendente a partire dal primo elemento.
Comportamento: il flusso produce elementi a prescindere dai subscriber. I nuovi subscriber vedono solo gli elementi emessi dopo la loro sottoscrizione.
La backpressure e un meccanismo fondamentale per gestire la differenza di velocita tra produttore e consumatore. Quando un produttore e molto piu veloce del consumatore, si crea un accumulo che puo portare a esaurimento della memoria. La backpressure permette al consumatore di "spingere all'indietro" un segnale che dice al produttore di rallentare.
La backpressure e il problema classico del produttore-consumatore: se il produttore produce piu velocemente di quanto il consumatore consumi, il buffer intermedio si riempie. Nei sistemi reattivi, questo problema emerge in modo naturale quando si compongono piu flussi con velocita diverse.
Il professore spiega che la backpressure trasforma il modello push puro in un modello push-pull ibrido: il downstream puo fare pull di N elementi se sono disponibili, ma se non lo sono, l'upstream li pusha quando prodotti.
// In Project Reactor: backpressure esplicita
Flux.range(1, 1000)
.subscribe(new BaseSubscriber<Integer>() {
@Override
protected void hookOnSubscribe(Subscription s) {
request(10); // chiede 10 elementi alla volta
}
@Override
protected void hookOnNext(Integer v) {
System.out.println(v);
request(1); // ne chiede 1 dopo ogni ricezione
}
});Il professore fa anche un esempio concreto con il metodo sample di Rx, che campiona un flusso ogni 500 millisecondi, poi fa un mapping su stringa e prende gli ultimi 5 elementi. Questo tipo di operatori sono estremamente utili nella pratica per filtrare flussi ad alta frequenza.
La backpressure non e solo un dettaglio tecnico: e fondamentale per costruire sistemi reattivi robusti. Senza backpressure, un produttore veloce puo far collassare il consumatore per out-of-memory. E per questo che specifiche come Reactive Streams (per JVM e JS) rendono la backpressure un requisito fondamentale.
La backpressure e uno dei concetti piu importanti della programmazione reattiva. Si verifica quando un produttore emette dati piu velocemente di quanto il consumatore riesca a processarli. Senza backpressure, i dati si accumulerebbero in buffer sempre piu grandi, portando a OutOfMemoryError o a un degrado delle prestazioni.
Il professore usa l'analogia della catena di montaggio presente nelle slide: la materia prima grezza fluisce da una sorgente (il Publisher), passa attraverso varie stazioni di lavoro (gli operatori), e arriva al prodotto finito che viene consegnato al consumatore (il Subscriber). Se una stazione di lavoro e piu lenta, deve poter segnalare a monte di rallentare: questo segnale di feedback e la backpressure.
flowchart LR
subgraph Produttore
P[Publisher: 1000 msg/s]
end
subgraph Buffer
B[(Buffer)]
end
subgraph Consumatore
C[Subscriber: 10 msg/s]
end
P -- overflow! --> B -- 10 msg/s --> C
C -- "request(10)" --> B
B -- "backpressure signal" --> P
Il modello push-pull ibrido e la chiave: il downstream puo fare pull di n elementi dall'upstream se sono pronti, ma se non lo sono, verranno spinti (push) dall'upstream appena prodotti. A differenza del modello Observable standard (senza backpressure), RxJava introduce Flowable che supporta nativamente la backpressure.
Ricordate: Observable NON supporta backpressure. Se avete bisogno di controllo del flusso, usate Flowable. Questa distinzione e fondamentale e viene spesso chiesta in sede d'esame, come sottolinea il professore.
ReactiveX offre diverse strategie per gestire la backpressure, ciascuna con caratteristiche specifiche:
| Strategia | Operatore | Comportamento |
|---|---|---|
| Buffering | buffer(size) | Raccoglie elementi in lotti di dimensione fissa e li emette come blocchi |
| Batching | window(size) | Come buffer, ma emette sotto-Observable invece di liste |
| Skipping | sample(duration) | Prende l'ultimo elemento a intervalli regolari, scartando il resto |
| Debouncing | debounce(duration) | Scarta elementi che arrivano troppo ravvicinati nel tempo |
| Throttling | throttleFirst(duration) | Prende il primo elemento di ogni finestra temporale |
Quando il buffer si riempie, si puo specificare una strategia di overflow tramite onBackpressureBuffer(capacity, callback, strategy):
ON_OVERFLOW_ERROR — lancia un'eccezioneON_OVERFLOW_DROP_LATEST — elimina l'elemento piu recente se il buffer e pienoON_OVERFLOW_DROP_OLDEST — elimina l'elemento piu vecchio per fare spazio al nuovoOppure si puo usare onBackpressureDrop() che semplicemente scarta gli elementi in eccesso senza buffer.
Nella programmazione reattiva ci si trova spesso a dover integrare codice esistente che non e reattivo. Il PublishSubject (o piu in generale i Subject) e la soluzione a questo problema.
Un Subject e un particolare tipo di flusso che funge sia da Observable (ci si puo iscrivere) che da Observer (ci si possono inviare dati manualmente). Il professore lo descrive come un canale che fornisce un'API per produrre valori (onNext, onComplete, onError) e consumarli (via subscribe).
PublishSubject<Integer> subject = PublishSubject.create();
// Mi iscrivo al subject come a un qualsiasi Observable
subject.subscribe(v -> System.out.println("Ricevuto: " + v));
// Produco valori manualmente, fuori dalla lambda di creazione!
subject.onNext(1);
subject.onNext(2);
subject.onComplete();La differenza fondamentale rispetto a Observable.create() e che qui gli onNext vengono chiamati dopo la creazione del flusso, da qualsiasi thread e da qualsiasi parte del codice. Il professore spiega che questo e il meccanismo ideale per "incollare" codice legacy (callback, listener, API sincrone) con codice reattivo.
Il PublishSubject crea un flusso hot: gli elementi emessi prima di una subscribe vengono persi (a meno di usare ReplaySubject che bufferizza e riemette tutta la cronologia). Scegliete il Subject giusto in base alle esigenze: PublishSubject per eventi in tempo reale, BehaviorSubject per l'ultimo valore, ReplaySubject per la cronologia completa.
Il professore mostra un esempio concreto (dal test 04A) di come agganciare i flussi reattivi a una GUI. L'esempio crea un pulsante e converte i suoi eventi di click in un flusso reattivo, permettendo di applicare operatori come map, filter, debounce e throttle agli eventi UI.
Il pattern tipico e:
onNext a ogni click// Creo il ponte tra GUI e mondo reattivo
PublishSubject<Void> onClickSubject = PublishSubject.create();
// Collego la GUI
button.addActionListener(e -> onClickSubject.onNext(null));
// Applico operatori come su qualsiasi flusso
onClickSubject
.debounce(300, TimeUnit.MILLISECONDS) // anti-rimbalzo
.observeOn(Schedulers.single()) // thread UI
.subscribe(e -> System.out.println("Click processato!"));Questo approccio, nota il professore, permette di separare completamente la logica di business dalla presentazione: la GUI si occupa solo di generare eventi, e il flusso reattivo si occupa di trasformarli, filtrarli e processarli.
Questa e forse la sezione piu importante per l'assignment del corso. Il professore mostra come flatMap + subscribeOn permetta di distribuire computazioni indipendenti su thread multipli in modo dichiarativo.
Flowable.range(1, 10)
.flatMap(v -> Flowable.just(v * v)
.subscribeOn(Schedulers.computation()))
.subscribe(v -> System.out.println(
Thread.currentThread().getName() + ": " + v));Quando si esegue questo codice, il professore fa notare che:
Il meccanismo e potente: invece di gestire manualmente un pool di thread e una coda di risultati, si dichiara semplicemente che ogni sottoflusso deve girare su un thread di computation, e RxJava si occupa di tutto il resto.
Questo pattern flatMap + subscribeOn e centrale nel secondo assignment. Il professore lo menziona esplicitamente: "Lo trovate in quello che chiamiamo test parallelism, vedrete che le varie map, 1, 2, 3, non sono piu in ordine, perche vengono servite da thread diversi." Saper spiegare perche l'ordine cambia e perche questo e un vantaggio e una conoscenza richiesta.
Il professore dedica una parte della lezione alla gestione dei thread in Vert.x, un framework reattivo per la JVM. Spiega un punto spesso frainteso: contrariamente a quanto si possa pensare, ogni Verticle ha un solo background thread, non un pool dinamico.
Quando si usa executeBlocking() in Vert.x, le computazioni bloccanti vengono accodate allo stesso background thread, non eseguite in parallelo. Il professore lo dimostra con un esempio concreto:
// Immaginate di chiamare executeBlocking due volte
vertx.executeBlocking(promise -> {
// blocking computation started - aspetta 5 secondi
Thread.sleep(5000);
promise.complete(100);
});
vertx.executeBlocking(promise -> {
// blocking computation started - deve aspettare che il primo finisca!
Thread.sleep(5000);
promise.complete(200);
});Anche se i due compiti sono indipendenti, vengono eseguiti in sequenza sullo stesso background thread: il secondo deve attendere che il primo sia completato. Il professore specifica: "Se vi mettete tanta roba da fare in background, vi rendete conto che non la fa in parallelo. Ha sempre un solo background thread, quindi li accoda."
Questa caratteristica ha implicazioni importanti per la progettazione: se si hanno piu task bloccanti indipendenti, conviene distribuirli su Verticle distinti o usare un pool di thread esterno gestito manualmente.
I thread pool di Vert.x non vanno confusi con gli Schedulers di RxJava. In Vert.x, l'event loop e unico per Verticle e l'executeBlocking accoda i task su un singolo background thread. RxJava, invece, offre Schedulers con pool di dimensioni variabili. Sono due astrazioni diverse che operano a livelli diversi.
Il professore chiude la lezione con alcune considerazioni importanti. La programmazione reattiva e le Reactive Extensions non sono un "silver bullet" per la programmazione asincrona. Sono efficaci per gestire flussi asincroni di dati/eventi in stile funzionale, ma non come modello di programmazione asincrona general-purpose.
La sfida attuale, dice il professore, e come integrare tutti questi approcci e tecniche:
La programmazione reattiva distribuita e un ambito di ricerca attivo: la gestione dei glitch (inconsistenze temporanee durante la propagazione del cambiamento) in contesto distribuito e ancora una questione aperta a causa di latenze di rete, assenza di clock globale e possibili guasti. I riferimenti a Salvaneschi et al. 2014 e Mogk et al. 2018 sono i punti di partenza su questo tema.
flatMap. Mentre map trasforma ogni elemento in un nuovo valore, flatMap trasforma ogni elemento in un Observable/Flowable e poi appiattisce tutti i sottoflussi in un unico flusso di output.
subscribeOn specifica il thread su cui viene eseguita la sorgente (a monte nella catena). observeOn specifica il thread su cui vengono eseguiti i subscriber e le operazioni a valle. subscribeOn ha effetto solo la prima volta che viene chiamato; per cambiare thread a meta catena serve observeOn.
Serve la backpressure. Con un Flowable (che supporta backpressure), il subscriber puo fare request(n) per segnalare quanti elementi e pronto a processare. Strategie come buffer, sample, debounce o onBackpressureDrop permettono di gestire il sovraccarico. Con un semplice Observable (senza backpressure), si rischia un OutOfMemoryError per buffer overflow.
Uno solo. Contrariamente a quanto si potrebbe pensare, Vert.x ha un singolo background thread per Verticle. Le chiamate a executeBlocking vengono accodate ed eseguite in sequenza, non in parallelo. Per parallelizzare task bloccanti, bisogna usare Verticle distinti o un pool di thread esterno.
Un Cold Observable inizia a emettere solo quando arriva un subscriber e ogni subscriber riceve tutti gli elementi dall'inizio (modello pull/lazy). Un Hot Observable emette continuamente indipendentemente dai subscriber e i subscriber tardivi ricevono solo gli elementi successivi alla loro iscrizione (modello push). Gli Hot Observable violano la regola "nothing happens until you subscribe".
I glitch sono inconsistenze temporanee che si verificano durante la propagazione del cambiamento: una computazione viene eseguita prima che tutte le sue dipendenze siano state aggiornate, combinando valori freschi con valori ancora obsoleti. Un esempio classico: se var2 = var1 * 1 e var3 = var1 + var2, cambiando var1 da 1 a 2, var3 potrebbe diventare momentaneamente 3 (se var3 viene ricalcolato prima di var2) prima di assestarsi a 4. I sistemi reattivi moderni evitano i glitch in contesto single-machine, ma in ambito distribuito il problema e ancora aperto.
Observable NON supporta backpressure. Flowable supporta la backpressure nativamente, permettendo al consumatore di segnalare al produttore di rallentare tramite request(n). Se avete bisogno di controllo del flusso (cosa comune in scenari con produttori veloci e consumatori lenti), usate Flowable.
Perche flatMap crea un nuovo flusso per ogni elemento in ingresso e subscribeOn su Schedulers.computation() fa eseguire ciascun sottoflusso su un thread diverso del pool di computation. I risultati vengono poi riaggregati automaticamente in un unico flusso di output. L'effetto e che elementi diversi vengono processati in parallelo su core diversi.