2. Un esempio introduttivo

I miei primi contatti con RxJava sono avvenuti attraverso corsi e tutorial trovati su Internet. Oltre al fatto che la teoria utilizzava concetti a cui non ero abituato e che facevo fatica a comprendere, soprattutto non riuscivo a capire a cosa potesse servire nella vita reale. Inizieremo quindi presentando un esempio (spero semplice) in cui l’uso di RxJava comporta una reale semplificazione nella scrittura del codice e, partendo da lì, cercheremo di individuare gli elementi importanti di questa libreria.

La libreria RxJava si basa sul seguente concetto: un flusso di elementi di tipo T Observable<T> viene osservato da uno o più sottoscrittori (abbonati, osservatori, consumatori) Subscriber<T>. La libreria RxJava consente al flusso Observable<T> di essere eseguito in un thread T1 e al suo osservatore Subscriber<T> in un thread T2 senza che lo sviluppatoredebba preoccuparsi di gestire il ciclo di vita di questi thread e di problemi naturalmente complessi, come la condivisione dei dati tra thread e la loro sincronizzazione per l’esecuzione di un’attività globale. Facilita quindi la programmazione asincrona.

Un flusso Observable<T> produce elementi di tipo T, osservabili man mano che vengono generati. Se l’osservatore e l’osservabile (termine che indica impropriamente il tipo Observable<T>) si trovano nello stesso thread, allora l’osservabile può produrre l’elemento (i+1) solo quando l’osservatore ha consumato l’elemento i. Sono pochi i casi in cui questa architettura risulta vantaggiosa. Se l’osservatore e l’osservabile non si trovano nello stesso thread, allora l’osservabile e il suo osservatore hanno comportamenti autonomi: l’osservabile produce al proprio ritmo e l’osservatore consuma al proprio ritmo. È proprio qui che risiede l’interesse della libreria. Finora abbiamo sempre parlato di un unico osservatore. In realtà, un osservabile può avere un numero qualsiasi di osservatori.

2.1. L’architettura dell’applicazione di esempio

L'applicazione di esempio presenta la seguente architettura:

in [1], un livello di servizio fornisce elenchi di numeri casuali. Questo livello viene eseguito nello stesso thread del metodo [swing] che lo utilizza. Fornisce quindi i suoi numeri in modo sincrono;
in [2], un sottile livello di adattamento implementato con RxJava consente di presentare al livello [swing] un’implementazione asincrona dello stesso servizio: quest’ultima può essere eseguita in un thread diverso da quello del metodo [swing] che la utilizza;
la chiamata [4] è sincrona, mentre la chiamata [5-6] è asincrona;

Ciò che vogliamo dimostrare qui è che la libreria Rx consente di trasformare facilmente un'interfaccia sincrona in un'interfaccia asincrona. Perché è utile? Gli eventi di un'interfaccia Swing vengono gestiti in un thread comunemente chiamato «event loop». Gli eventi vengono messi in coda e gestiti uno dopo l'altro. L’evento Ei+1 può essere elaborato solo quando l’evento precedente Ei è stato completamente elaborato. È quindi importante che l’elaborazione di un evento sia il più breve possibile affinché l’interfaccia grafica rimanga reattiva. A volte, l’elaborazione di un evento può richiedere molto tempo. Ciò accade, ad esempio, se tale elaborazione comporta accessi di rete. Se non si vuole bloccare l’interfaccia grafica in modo inaccettabile per l’utente, è necessario che tali accessi di rete avvengano in thread separati dal ciclo di eventi, in modo da liberare quest’ultimo. Ci si addentra quindi nel campo della programmazione concorrente (più thread vengono eseguiti in parallelo), giustamente considerata difficile. La libreria Rx offre una soluzione semplice ed elegante a questo problema.

Per simulare elaborazioni lunghe, il servizio dell’esempio fornisce i propri numeri casuali dopo un certo tempo di attesa, in modo da poter osservare il comportamento dell’interfaccia grafica.

2.2. L'exécutable

Il file eseguibile dell’applicazione di esempio si trova nella cartella [dvp/executables] degli esempi:

Esistono diversi modi per eseguire l’archivio [swing-01] a seconda della configurazione del computer utilizzato per l’esecuzione. Ad esempio, è possibile seguire la procedura descritta in [1-3]. Si otterrà quindi la seguente interfaccia grafica:

l'interfaccia presenta due schede [1-2]: una, [Request], per la richiesta al servizio di generazione di numeri casuali, l'altra, [Response], per la visualizzazione dei numeri ricevuti;
in [3] si indica il numero di richieste che si desidera inviare al servizio;
in [4] si indica l’intervallo [a,b] di generazione dei numeri desiderati;
in [5], il numero di valori restituiti dal servizio sarà un numero casuale compreso nell'intervallo [minCount, maxCount] stabilito dall'utente;
in [6], prima di restituire la risposta, il servizio attenderà delay millisecondi, dove delay è un numero casuale compreso nell’intervallo [minDelay, maxDelay] definito dall’utente;
per impostazione predefinita, il livello [swing] si rivolgerà all’interfaccia sincrona del servizio. Per rivolgersi al livello asincrono, l’utente dovrà selezionare [7]. In questo caso, il servizio di generazione verrà eseguito in thread separati dall’event loop dell’interfaccia grafica. La libreria Rx dispone di diverse strategie per la generazione di questi thread. L’utente potrà scegliere la propria strategia in [8];
la generazione dei numeri avviene tramite il pulsante [9];

in [10], visualizzazione dei risultati. Spiegheremo la struttura di questi ultimi;
in [11], il numero di risultati ottenuti;
in [12], il tempo di esecuzione in millisecondi;
in [13], l'utente ha la possibilità di annullare l'esecuzione;

Ogni risultato ha la seguente forma:

{"idClient":0,"serviceResponse":{"delay":412,"aleas":[146,115,128,174,159,112,162,127],"executedOn":"RxComputationThreadPool-6"},"observedOn":"AWT-EventQueue-0","requestAt":"02:42:47:708","responseAt":"02:42:52:931"}

[idClient]: il numero della richiesta. Si ricorda che vengono effettuate più richieste al servizio di generazione;
[delay]: il tempo di attesa in millisecondi che il servizio ha osservato prima di inviare il proprio risultato;
[aleas]: i numeri casuali restituiti dal servizio;
[executedOn]: il nome del thread in cui è stato eseguito il servizio;
[observedOn]: il nome del thread che ha visualizzato il risultato. Con un'interfaccia Swing, questo non può che essere il thread dell'event loop, in questo caso [AWT-EventQueue-0];
[requestAt]: l'ora della richiesta nel formato [heures:minutes:secondes:millisecondes];
[responseAt]: l'ora di ricezione dei risultati nello stesso formato;

Presenteremo ora le parti di codice utili per la comprensione dell'esempio.

2.3. L'interfaccia sincrona

Il livello di servizio [1] presenta la seguente interfaccia:


public interface IService {
  // numeri casuali in [a,b]
  // vengono generati n numeri con n casuali nell'intervallo [minCount, maxCount]
  // i numeri vengono generati dopo un ritardo di delay millisecondi,
  // dove [delay] è un numero casuale nell'intervallo [minDelay, maxDelay]
  public ServiceResponse getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay);
}

La risposta [ServiceResponse] è la seguente:


public class ServiceResponse {

  // tempo di attesa del servizio
  private int delay;
  // numeri casuali
  private List<Integer> aleas;
  // thread di esecuzione
  private String executedOn;

  // costruttori

  public ServiceResponse(int delay, List<Integer> aleas) {
    executedOn = Thread.currentThread().getName();
    this.delay = delay;
    this.aleas = aleas;
  }

  // getter e setter
...
}

La risposta è composta da tre elementi:

riga 6: i numeri casuali generati;
riga 4: il tempo di attesa osservato dal servizio prima di restituire il risultato;
riga 8: il thread di esecuzione del servizio;

2.4. La chiamata sincrona

Descriviamo ora in dettaglio la chiamata sincrona [4] che il livello [swing] effettua al servizio [1]:


  private void doGenerateWithService() {
    // inizio attesa
    beginWaiting();
    try {
      for (int i = 0; i < nbRequests; i++) {
        UiResponse uiResponse = new UiResponse();
        uiResponse.setIdClient(i);
        uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
        uiResponse.setResponseAt();
        model.add(0, jsonMapper.writeValueAsString(uiResponse));
        jLabelNbReponses.setText(String.valueOf(Integer.parseInt(jLabelNbReponses.getText()) + 1));
      }
    } catch (JsonProcessingException | RuntimeException e) {
      System.out.println(e);
    }
    // fine attesa
    endWaiting();
}

righe 5-12: il ciclo di esecuzione delle richieste [nbRequests] inviate dall’utente;
riga 8: [service] è l’implementazione dell’interfaccia sincrona [IService] descritta nel paragrafo 2.3;
riga 10: [model] è il modello visualizzato dal componente JList della scheda [Response]. Gli elementi di questo modello sono le stringhe jSON dei seguenti elementi di tipo [UiResponse]:


public class UiResponse {

  // ID del cliente
  private int idClient;
  // risposta del servizio
  private ServiceResponse serviceResponse;
  // nome del thread di osservazione
  private String observedOn;
  // ora della richiesta
  private String requestAt;
  // ora della risposta
  private String responseAt;

  // costruttori

  public UiResponse() {
    observedOn = Thread.currentThread().getName();
    requestAt = getTimeStamp();
  }
  // metodi privati

  private String getTimeStamp() {
    return new SimpleDateFormat("hh:mm:ss:SSS").format(Calendar.getInstance().getTime());
  }

  // getter e setter
...
}

riga 6: la risposta del servizio di generazione dei numeri;
riga 4: il numero della richiesta a cui si risponde;
riga 8: il thread di visualizzazione di questa risposta. Come già detto, sarà sempre il thread dell'event loop;
righe 10 e 12: l'ora della richiesta e quella della risposta;

2.5. Test delle chiamate sincrone

Eseguiamo la seguente configurazione:

Otteniamo i seguenti risultati nella scheda [Response]:

in [1-2], abbiamo effettivamente ottenuto 10 risposte come richiesto. Sono state inserite in prima posizione nell'ordine di arrivo. Si nota che sono state ottenute nell'ordine delle richieste;
sono state tutte eseguite e visualizzate nel thread dell’event loop [AWT-EventQueue-0]. Le richieste sono state quindi eseguite una dopo l’altra in questo thread. Non ci sono state richieste simultanee;
ciò che qui non è visibile è che durante l’esecuzione l’interfaccia grafica è bloccata. Ad esempio, non è possibile accedere alla scheda [Response] per vedere le risposte in arrivo o interrompere l’esecuzione con il pulsante [Annuler]. Anche se questo pulsante fosse stato presente nella scheda [Request], sarebbe stato inutilizzabile. Infatti, si sarebbero verificati due eventi:
- il clic sul pulsante [Générer];
- il clic sul pulsante [Annuler];

Il clic sul pulsante [Annuler] viene gestito solo al termine dell’operazione avviata dal clic sul pulsante [Générer]. Abbiamo appena visto che quest'ultima occupava il thread dell'event loop per tutta la durata dell'esecuzione, impedendo così la gestione del clic sul pulsante [Annuler]. Si tratta tipicamente del tipo di situazioni in cui Rx può apportare un netto miglioramento;

2.6. L’interfaccia asincrona e la sua implementazione

Ci concentriamo ora sull’interfaccia del livello [2] e sulla sua implementazione con Rx. Quest’ultima non sarà immediatamente comprensibile. Vogliamo semplicemente mettere in evidenza la semplicità del codice di questa implementazione.

L'interfaccia asincrona è la seguente:


public interface IRxService {
  // numeri casuali in [a,b]
  // vengono generati n numeri con n casuale nell'intervallo [minCount, maxCount]
  // i numeri vengono generati dopo un ritardo di delay millisecondi,
  // dove [delay] è un numero casuale nell'intervallo [minDelay, maxDelay]
  public Observable<UiResponse> getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay, UiResponse uiResponse);
}

Le differenze rispetto all’interfaccia sincrona presentata nel paragrafo 2.3 sono le seguenti:

la classe [UiResponse] presentata al paragrafo 2.3 fa ora parte dei parametri del metodo [getAleas] (riga 6). Il motivo è che, poiché le richieste vengono ora eseguite in parallelo e il servizio attende un tempo casuale prima di restituire il risultato, le risposte non ci perverranno nell’ordine delle richieste. Si passa quindi l’oggetto [UiResponse] che contiene, tra le altre informazioni, il numero della richiesta:


  // ID del cliente (richiesta)
  private int idClient;
  // risposta del servizio
  private ServiceResponse serviceResponse;
  // nome del thread di osservazione
  private String observedOn;
  // ora della richiesta
  private String requestAt;
  // ora della risposta
  private String responseAt;

Il tipo della risposta del servizio asincrono è [Observable<UiResponse>]. Il tipo [Observable<>] è fornito dalla libreria Rx. Il risultato di tipo [Observable<UiResponse>] indica che il metodo [getAleas] fornisce un flusso di valori di tipo [UiResponse], valori che vengono inviati (pushed) uno alla volta al proprio osservatore;

Vediamo ora l’implementazione di questa interfaccia:


public class RxService implements IRxService {

  // servizio
  private IService service;

  // costruttore
  public RxService(IService service) {
    this.service = service;
  }

  @Override
  public Observable<UiResponse> getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay, UiResponse uiResponse) {
    return Observable.create(subscriber -> {
      try {
        uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
        subscriber.onNext(uiResponse);
      } catch (Exception e) {
        subscriber.onError(e);
      } finally {
        subscriber.onCompleted();
      }
    });
  }
}

righe 7-9: al costruttore viene fornito un riferimento all’interfaccia sincrona [IService]. Sarà questa a occuparsi della generazione dei numeri casuali;
l’osservabile restituito dal metodo [getAleas] viene costruito dal metodo statico [Observable.create]. È questo metodo che permette di costruire un’implementazione asincrona a partire da un’implementazione sincrona;
riga 13: il parametro del metodo statico [Observable.create] è qui una funzione lambda che riceve come parametro un tipo [Subscriber], anch’esso un tipo Rx. Un [Subscriber] è un oggetto che si abbona a un flusso di osservabili, ovvero un flusso di dati forniti in modo asincrono. Qui si utilizzano tre metodi di questo abbonato:
- [Subscriber.onNext] per trasmettergli un dato (riga 16);
- [Subscriber.onError] per trasmettergli un'eccezione (riga 18);
- [Subscriber.onCompleted] per indicare all’abbonato che il flusso di dati è terminato (riga 20);

Possono esserci più abbonati allo stesso osservabile. In questo caso, avremo un solo abbonato che si abbona a un flusso di un unico dato, quello generato alle righe 15-16. Il dato viene generato dall’implementazione sincrona del servizio (riga 15) e restituito all’abbonato (riga 16).

Anche se tutto ciò rimane probabilmente oscuro, non si può non rimanere colpiti dall’estrema concisione di questa implementazione asincrona del servizio.

2.7. La chiamata asincrona

Analizziamo ora in dettaglio la chiamata sincrona [5] che il livello [swing] effettua al servizio [2]:


private void doGenerateWithRxService() {
        // inizio attesa
        beginWaiting();
        // richiesta di numeri casuali
        Observable<UiResponse> observables = Observable.empty();
        for (int i = 0; i < nbRequests; i++) {
            UiResponse uiResponse = new UiResponse();
            uiResponse.setIdClient(i);
            // scheduler
            int schedulerIndex = jComboBoxSchedulers.getSelectedIndex();
            switch (schedulerIndex) {
            case 0:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.io()));
                break;
...
            }
        }
...
    }

righe 6-10: esecuzione delle richieste [nbRequests] effettuate dall’utente;
righe 7-8: preparazione dell’oggetto [UiResponse] necessario al metodo [getAleas] del servizio asincrono (riga 13). Si tratta principalmente di registrare il numero [idClient] della richiesta;
riga 13: viene chiamato il metodo [getAleas] del servizio asincrono. Esso restituisce un oggetto [Observable<UiResponse>]. Questa chiamata non invoca ancora il servizio sincrono. Torniamo al codice del metodo asincrono [getAleas]:


  @Override
  public Observable<UiResponse> getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay, UiResponse uiResponse) {
    return Observable.create(subscriber -> {
      try {
        uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
        subscriber.onNext(uiResponse);
      } catch (Exception e) {
        subscriber.onError(e);
      } finally {
        subscriber.onCompleted();
      }
    });
}

Il codice delle righe 4-11, che richiamerà il servizio sincrono, viene eseguito solo quando un abbonato si registra. Finché non ci sono abbonati, questo codice non viene eseguito.

Torniamo al codice del metodo [doGenerateWithRxService]:

riga 5: si crea un osservabile vuoto (non viene osservato nulla);
riga 13: si crea un osservabile il cui flusso sarà la fusione dei flussi asincroni [nbRequests] associati alle richieste [nbRequests]. Ciò si ottiene con il metodo [Observable.mergeWith], che consente di fondere due flussi asincroni. Nella terminologia Rx, [mergeWith] è definito un operatore di flusso. Questi operatori hanno la particolarità che il risultato dell’operazione è, nella maggior parte dei casi, nuovamente un [Observable]. Alla fine, dopo la riga 17, la variabile [observables] indica un unico flusso costituito dalle risposte asincrone [nbRequests] fornite dal servizio asincrono;
riga 13: l’operazione di fusione avrebbe potuto essere scritta come segue:


observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse));

ma abbiamo scritto:


observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.io()));

Qui abbiamo utilizzato l’operatore [subscribeOn] sull’osservabile [rxService.getAleas]. Come spesso accade, il risultato è nuovamente un osservabile. L’operatore [subscribeOn] consente di specificare che l’osservabile deve essere eseguito in un thread fornito da un [Scheduler]. Esistono diversi [Scheduler] possibili, adatti a diverse situazioni. Nell’interfaccia grafica ne abbiamo proposti diversi per vedere gli effetti di ciascuno:

Questo genera il seguente codice:


    private void doGenerateWithRxService() {
        // inizio attesa
        beginWaiting();
        // richiesta di numeri casuali
        Observable<UiResponse> observables = Observable.empty();
        for (int i = 0; i < nbRequests; i++) {
            UiResponse uiResponse = new UiResponse();
            uiResponse.setIdClient(i);
            // scheduler
            int schedulerIndex = jComboBoxSchedulers.getSelectedIndex();
            switch (schedulerIndex) {
            case 0:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.io()));
                break;
            case 1:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.computation()));
                break;
            case 2:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.newThread()));
                break;
            case 3:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.trampoline()));
                break;
            case 4:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.immediate()));
                break;
            }
        }
...
}

Torniamo al codice delle righe 12-14. Lo scheduler [Schedulers.io()] assegna un nuovo thread a ciascun osservabile. Se seguiamo il codice:

riga 5: abbiamo un osservabile vuoto;
riga 13, iterazione 1: observables è la lista [observable0/thread0] (osservabile observable0 eseguito sul thread thread0);
riga 13, iterazione 2: observables è la lista [observable0/thread0, observable1/thread1];
ecc...

Alla fine, dopo la riga 28, si ottiene un osservabile risultante dalla fusione degli osservabili [nbRequests] che vengono eseguiti su thread [nbRequests] diversi. Non tutti gli scheduler funzionano in questo modo, come vedremo durante i test.

Continuiamo l’analisi del codice di chiamata del servizio asincrono:


private void doGenerateWithRxService() {
        // inizio attesa
        beginWaiting();
        // richiesta di numeri casuali
        Observable<UiResponse> observables = Observable.empty();
        for (int i = 0; i < nbRequests; i++) {
        ...
        }
        // osservatore
        observables = observables.observeOn(SwingScheduler.getInstance());
        // si eseguono questi osservabili
        subscriptions.add(observables.subscribe(uiResponse -> {
            updateUi(uiResponse);
        } , th -> {
            System.out.println(th);
            doCancel();
        } , this::doCancel));
    }

abbiamo visto che quando si arriva alla riga 10, si ha un unico osservabile, risultato della fusione di [nbRequests] osservabili che possono essere eseguiti su [nbRequests] thread diversi o meno, a seconda dello scheduler scelto dall’utente;
riga 10: l’operatore [observeOn] consente di specificare su quale thread si desidera recuperare i dati provenienti dall’osservabile, in questo caso gli oggetti di tipo [UiResponse]. In un'interfaccia Swing non si ha scelta. Qualsiasi aggiornamento dell'interfaccia deve avvenire nel thread dell'event loop. In questo caso, i dati dell'observable verranno visualizzati in un componente Swing JList. Il thread [SwingScheduler.getInstance()] rappresenta il thread dell’event loop. La classe [SwingScheduler] non proviene dalla libreria RxJava ma da quella derivata RxSwing;
quando si arriva alla riga 12, il servizio sincrono non è ancora stato chiamato poiché l’osservabile della riga 10 non ha ancora alcun sottoscrittore. Le righe 12-17 ne assegnano uno, grazie all’operatore [subscribe]. I parametri di questo operatore sono qui tre funzioni lambda:
- la prima, [uiResponse -> {updateUi(uiResponse);}], accetta come parametro uno degli oggetti [UiResponse] generati dall’osservabile. Ricordiamo che in questo caso avremo [nbRequests] oggetti di questo tipo. Il metodo associato, in questo caso updateUi, deve utilizzare questo risultato;
- Il secondo [th -> {System.out.println(th);doCancel();}] accetta come parametro un tipo [Throwable], in questo caso un'eccezione verificatasi durante l'esecuzione dell'osservabile. Il metodo associato deve utilizzare questa informazione. In questo caso, l’informazione viene visualizzata sulla console (riga 15) e l’esecuzione viene interrotta, il che comporterà l’aggiornamento di alcuni elementi dell’interfaccia grafica;
- il terzo [this::doCancel] viene chiamato quando l’osservabile segnala di non avere più dati da trasmettere. In questo caso, l’osservabile è l’unione degli osservabili [nbRequests]. L’observable risultante indicherà di aver terminato quando tutti gli observable che lo compongono avranno a loro volta segnalato di aver completato il proprio lavoro. Pertanto, quando viene eseguita questa terza funzione lambda, tutti i dati sono stati ricevuti. Il metodo locale [doCancel] aggiorna l’interfaccia grafica per riflettere il fatto che l’esecuzione è terminata;

La variabile [subscriptions] è definita come segue:


    // le sottoscrizioni agli osservabili
protected List<Subscription> subscriptions = new ArrayList<Subscription>();

Il tipo [Subscription] rappresenta un abbonamento, ovvero il collegamento tra un abbonato [Subscriber] e ciò che osserva [Observable]. Qui abbiamo utilizzato un elenco di abbonamenti, sebbene in questo esempio ce ne sia solo uno. Il metodo locale [doCancel], eseguito quando l’osservabile segnala di non avere più dati da trasmettere, è il seguente:


    @Override
    protected void doCancel() {
        // fine attesa
        endWaiting();
        // in caso di sottoscrizioni
        if (jCheckBoxRxSwing.isSelected() && subscriptions != null) {
            subscriptions.forEach(Subscription::unsubscribe);
        }
}

la riga 7 cancella l'iscrizione di tutti gli iscritti all'osservabile;

Da questa breve spiegazione si possono trarre i seguenti punti chiave:

il tipo [Observable] indica un flusso di valori, valori che vengono inviati uno alla volta agli abbonati o agli osservatori;
il tipo [Subscriber] indica un abbonato del tipo [Observable];
il tipo [Subscription] indica un abbonamento, ovvero il collegamento tra un [Subscriber] e un [Observable];
il tipo [Observable] ammette operatori [mergeWith, empty, subscribeOn, observeOn, ...] che, nella maggior parte dei casi, generano osservabili. Questi operatori servono a configurare l'osservabile prima della sua esecuzione:
- ciò che si desidera osservare;
- il thread su cui viene eseguito l’osservabile;
- il thread su cui l’abbonato riceve i dati dall’osservabile;
si distinguono due tipi di osservabili: i [froid / cold] e i [chaud / hot]. Un osservabile freddo viene eseguito interamente ogni volta che si aggiunge un nuovo abbonato. Se ogni esecuzione produce gli stessi dati, ogni nuovo abbonato riceve gli stessi dati del precedente. Un osservabile "caldo" produce generalmente dati in modo continuo. Quando un sottoscrittore si abbona, riceve i dati emessi a partire dal momento della sua sottoscrizione. Non riceve i dati che potrebbero essere stati emessi in precedenza. Nel nostro esempio, l’osservabile è freddo: viene rieseguito interamente ogni volta che si aggiunge un nuovo sottoscrittore. Cosa viene effettivamente eseguito nel nostro esempio? Per scoprirlo, occorre tornare alla definizione dell’osservabile osservato:


  @Override
  public Observable<UiResponse> getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay, UiResponse uiResponse) {
    return Observable.create(subscriber -> {
      try {
        uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
        subscriber.onNext(uiResponse);
      } catch (Exception e) {
        subscriber.onError(e);
      } finally {
        subscriber.onCompleted();
      }
    });
}

Ad ogni nuovo sottoscrittore, la funzione lambda, parametro del metodo [Observable.create] (riga 3), viene rieseguita. Sono quindi le righe 4-11 che vengono eseguite per ogni nuovo sottoscrittore [subscriber];

2.8. Test delle chiamate asincrone

Iniziamo illustrando l’effetto dei diversi scheduler disponibili. A tal fine utilizziamo i seguenti parametri:

In [1-2] impostiamo valori bassi in modo che, se le richieste vengono eseguite sullo stesso thread, l’attesa non sia comunque troppo lunga.

2.8.1. con lo scheduler [Schedulers.io]

Si possono notare i seguenti punti:

le risposte vengono ricevute in un ordine diverso da quello delle richieste (cfr. idClient);
ogni richiesta è stata eseguita in un thread diverso;
l'interfaccia grafica questa volta non è più bloccata:
- è possibile passare da una scheda all’altra;
- si vedono arrivare i dati;
- non si ha il tempo di vedere il pulsante [Annuler] perché l'esecuzione è troppo veloce. Lo metteremo in evidenza in un altro test;

2.8.2. con lo scheduler [Schedulers.computation]

Si possono notare i seguenti punti:

le risposte vengono ricevute in un ordine diverso da quello delle richieste (cfr. idClient);
le richieste sono state eseguite in 8 thread;
il thread n. 3 è stato utilizzato per le richieste 8 e 0;
il thread n. 4 è stato utilizzato per le richieste 9 e 1;
le altre richieste hanno avuto ciascuna un thread diverso;

Lo scheduler [Schedulers.computation] utilizza un numero di thread pari al numero di core presenti sulla macchina utilizzata. Questa informazione viene ottenuta tramite l'espressione [Runtime.getRuntime().availableProcessors()].

2.8.3. con lo scheduler [Schedulers.newThread]

Il funzionamento è analogo a quello dello scheduler [Schedulers.io].

2.8.4. con gli scheduler [Schedulers.trampoline, Schedulers.immediate]

Il funzionamento è sincrono. Tutte le richieste vengono eseguite sul thread dell’event loop. Non bisogna generalizzare questo risultato, ma semplicemente constatare che in questo specifico esempio i due scheduler hanno funzionato in modo sincrono.

2.9. Casi limite

In questo esempio lavoreremo con gli scheduler che consentono un funzionamento asincrono. Innanzitutto aumentiamo il numero di richieste a 100 con lo scheduler [Schedulers.computation], che in questo caso opera con 8 thread. Otteniamo il seguente risultato:

in [1], il pulsante [Annuler] è presente e utilizzabile (funzionamento asincrono);

Ora lasciamo che l’esecuzione vada a buon fine:

Da [2] si evince che l'esecuzione delle 100 query ha richiesto circa 4 secondi (su 8 thread).

Ora eseguiamo queste stesse 100 richieste con lo scheduler [Schedulers.newThread], che esegue ogni richiesta su un thread separato:

In [1], si osserva che l’esecuzione delle 100 richieste (su 100 thread) ha richiesto mezzo secondo. È quindi nettamente più veloce rispetto allo scheduler [Schedulers.computation].

Ora eseguiamo 800 richieste nelle stesse condizioni, sempre con lo scheduler [Schedulers.newThread]. Si ottengono i seguenti risultati:

Le 800 richieste vengono eseguite in circa 1 secondo.

Quando si aumenta questo numero (oltre le 2500 richieste sul mio computer – eseguite in 1,5 s – questo numero dipende ovviamente molto dall’ambiente di lavoro al momento dell’esecuzione), si ottiene la seguente eccezione:

Si verifica quindi un overflow dello stack. I test dimostrano che il funzionamento dello scheduler [Schedulers.newThread] non è deterministico. È possibile che si verifichi l’eccezione precedente, effettuare nuovi tentativi, tornare poi alla configurazione che ha causato l’eccezione e non riscontrarla più.

2.10. Conclusion

Abbiamo illustrato un esempio di utilizzo della libreria Rx. Riassumiamo ciò che abbiamo appreso:

Siamo partiti dalla seguente architettura:

in [4], il livello [swing] effettuava chiamate sincrone al livello [service];
in [5], il livello [swing] effettuava chiamate asincrone al livello [rxService], che a sua volta chiamava in modo sincrono il livello [service] tramite [6];

La prima cosa che abbiamo notato è che la libreria Rx consentiva di creare facilmente l’interfaccia asincrona [rxService] a partire dall’interfaccia sincrona [service] (cfr. paragrafo 2.4). Si tratta di un insegnamento importante perché significa che è possibile far evolvere facilmente un’applicazione sincrona in un’applicazione asincrona.

Nel livello [swing] sono stati scritti due metodi distinti:

uno per effettuare chiamate sincrone al servizio (cfr. paragrafo 2.4);
l’altro per effettuare chiamate asincrone al servizio (cfr. paragrafo 2.7);

La scrittura delle chiamate asincrone si è rivelata nettamente più complessa rispetto a quella delle chiamate sincrone. Tuttavia, chi ha esperienza di programmazione concorrente con più thread da sincronizzare, troverà che la soluzione Rx risulta più semplice da implementare ed evita tutti i problemi di sincronizzazione e comunicazione tra thread, che sono particolarmente complessi. Durante la stesura di questo documento, abbiamo individuato i seguenti punti importanti:

il tipo [Observable] indica un flusso di eventi (valori) che possono essere (ma non necessariamente) asincroni e che possono essere osservati;
il tipo [Subscriber] indica un abbonato a un tipo [Observable];
il tipo [Subscription] indica un abbonamento, ovvero il collegamento tra un [Subscriber] e un [Observable];
il tipo [Observable] ammette operatori [mergeWith, empty, subscribeOn, observeOn, ...] che, per la maggior parte, producono osservabili. Questi operatori servono a configurare l'osservabile prima della sua esecuzione:
- ciò che si desidera osservare;
- il thread su cui viene eseguito l'osservabile;
- il thread su cui l’abbonato riceve i dati dall’osservabile;
si distinguono due tipi di osservabili: i [froid / cold] e i [chaud / hot]. Un osservabile freddo viene eseguito interamente ogni volta che si aggiunge un nuovo abbonato. Se ogni esecuzione produce gli stessi dati, ogni nuovo abbonato riceve gli stessi dati del precedente. Un osservabile "caldo" produce generalmente dati in modo continuo. Quando un sottoscrittore si abbona, riceve i dati emessi a partire dal momento della sua sottoscrizione. Non riceve i dati che potrebbero essere stati emessi in precedenza. Nel nostro esempio, l’osservabile è "freddo": viene rieseguito interamente ad ogni nuovo sottoscrittore.

Ora che abbiamo visto un esempio che ci ha illustrato l’utilità della libreria Rx, la presenteremo più nel dettaglio.

La libreria Rx dispone di numerosi metodi con parametri generici nella loro firma. Faremo un breve ripasso su queste firme (paragrafo 3). I parametri di questi metodi sono per lo più interfacce funzionali (Java 8), ovvero interfacce che hanno un unico metodo. I parametri effettivi devono quindi essere istanze di queste interfacce. Prima di Java 8, era consuetudine implementare un’interfaccia tramite una classe anonima. Con Java 8, se l’interfaccia è di tipo funzionale, risulta più conciso implementarla con una funzione lambda. Presenteremo quindi queste ultime (paragrafo 4). Una volta fatto ciò, presenteremo la classe [Stream] (paragrafo 5) che consente di elaborare le collezioni Java con le funzioni lambda. Questa classe è interessante perché la classe [Observable] di RxJava ne prende in prestito:

alcuni metodi;
lo stesso modo di concatenare i metodi tra loro per elaborare lo stesso osservabile;

Presenteremo quindi le interfacce funzionali specifiche della libreria RxJava (paragrafo 6). Proseguiremo con i principali elementi della libreria Rx [Observable, Subscriber, Subscription, opérateurs] (paragrafo 7). La classe [Observable] dispone di diverse decine di operatori che a loro volta sono sovraccaricati più volte. Ciò crea inizialmente una grande complessità poiché questi operatori e i loro sovraccarichi a volte differiscono solo per un dettaglio ed è difficile, senza esperienza, sapere quale operatore utilizzare. Presenteremo solo un numero limitato di operatori e, nella maggior parte dei casi, tralasceremo i loro sovraccarichi.

L’intera parte precedente verrà realizzata con la libreria RxJava in semplici applicazioni da console. Una volta acquisita la libreria RxJava, la utilizzeremo in due tipi di applicazioni grafiche:

nel paragrafo 8 torneremo sull’applicazione Swing di esempio per approfondirla ulteriormente. Utilizzeremo quindi la libreria RxSwing;
nel paragrafo 9 creeremo un’applicazione Android con la libreria RxAndroid;

Una volta completato tutto questo, il lettore avrà gli strumenti necessari per cavarsela da solo. Probabilmente ci vorrà del tempo prima che riesca a utilizzare la libreria Rx in modo intuitivo. Ho trovato questa libreria particolarmente interessante. Tuttavia, l’ho trovata complessa da comprendere e il periodo di apprendimento è stato lungo. Spero che questo documento consenta al lettore di accorciare tale periodo. Mi sembra che ne valga davvero la pena.