2. Een inleidend voorbeeld

Mijn eerste kennismaking met RxJava vond plaats via cursussen en tutorials die ik op internet had gevonden. Niet alleen maakte de theorie gebruik van concepten die ik niet gewend was en die ik moeilijk kon begrijpen, maar ik zag vooral niet in wat het nut ervan zou kunnen zijn in het echte leven. We zullen daarom beginnen met een (hopelijk eenvoudig) voorbeeld waarin het gebruik van RxJava het schrijven van code aanzienlijk vereenvoudigt, en van daaruit zullen we proberen de belangrijke elementen van deze bibliotheek in kaart te brengen.

De bibliotheek RxJava is gebaseerd op het volgende concept: een stroom van elementen van het type T Observable<T> wordt geobserveerd door een of meer abonnees (subscribers, observers, consumers) Subscriber<T>. De bibliotheek RxJava maakt het mogelijk dat de Observable<T>-stroom in een T1-thread wordt uitgevoerd en de bijbehorende waarnemer Subscriber<T> in een T2-thread, zonder dat de ontwikkelaarzich zorgen hoeft te maken over het beheer van de levenscyclus van deze threads en over van nature lastige problemen, zoals het delen van gegevens tussen threads en de synchronisatie daarvan om een globale taak uit te voeren. Het vergemakkelijkt dus asynchroon programmeren.

Een Observable<T>-stream produceert elementen van het type T, die kunnen worden geobserveerd zodra ze worden geproduceerd. Als de waarnemer en de observable (een verkeerde benaming voor het type Observable<T>) zich in dezelfde thread bevinden, dan kan de observable het element (i+1) pas produceren wanneer de waarnemer het element i heeft verwerkt. Er zijn maar weinig gevallen waarin deze architectuur zinvol is. Als de waarnemer en het observeerbare object zich niet in dezelfde thread bevinden, dan gedragen het observeerbare object en zijn waarnemer zich onafhankelijk van elkaar: het observeerbare object produceert in zijn eigen tempo en de waarnemer verbruikt in zijn eigen tempo. Daarin ligt het nut van de bibliotheek. Tot nu toe hebben we het altijd over één waarnemer gehad. In werkelijkheid kan een observeerbaar object een willekeurig aantal waarnemers hebben.

2.1. De architectuur van de voorbeeldtoepassing

De voorbeeldtoepassing heeft de volgende architectuur:

• in [1] levert een servicelaag lijsten met willekeurige getallen. Deze laag wordt uitgevoerd in dezelfde thread als de methode [swing] die er gebruik van maakt. Ze levert haar getallen dan synchroon;
• in [2] maakt een dunne adaptatielaag, geïmplementeerd met RxJava, het mogelijk om aan de laag [swing] een asynchrone implementatie van dezelfde service te presenteren: deze kan worden uitgevoerd in een andere thread dan die van de methode [swing] die er gebruik van maakt;
• de aanroep [4] is synchroon, terwijl de aanroep [5-6] juist asynchroon is;

Wat we hier willen laten zien, is dat de Rx-bibliotheek het mogelijk maakt om een synchrone interface eenvoudig om te zetten in een asynchrone interface. Waarom is dit nuttig? De gebeurtenissen van een Swing-interface worden verwerkt in een thread die gewoonlijk de ‘event loop’ wordt genoemd. De gebeurtenissen worden in een wachtrij geplaatst en één voor één verwerkt. Gebeurtenis Ei+1 kan pas worden verwerkt wanneer de voorgaande gebeurtenis Ei volledig is verwerkt. Het is daarom belangrijk dat de verwerking van een gebeurtenis zo kort mogelijk duurt, zodat de grafische interface responsief blijft. Soms kan de verwerking van een gebeurtenis veel tijd in beslag nemen. Dit is het geval wanneer deze verwerking netwerktoegang vereist. Als we willen voorkomen dat de grafische interface op een voor de gebruiker onaanvaardbare manier vastloopt, moeten deze netwerktoegangen plaatsvinden in threads die gescheiden zijn van de event loop, om deze vrij te maken. We komen dan terecht in het domein van concurrente programmering (waarbij meerdere threads parallel worden uitgevoerd), dat terecht als moeilijk wordt beschouwd. De Rx-bibliotheek biedt een eenvoudige en elegante oplossing voor dit probleem.

Om langdurige bewerkingen te simuleren, levert de service in het voorbeeld zijn willekeurige getallen pas na een bepaalde wachttijd af, zodat we het gedrag van de grafische interface kunnen zien.

2.2. L'exécutable

Het uitvoerbare bestand van de voorbeeldtoepassing bevindt zich in de map [dvp/executables] van de voorbeelden:

Er zijn verschillende manieren om het archief [swing-01] uit te voeren, afhankelijk van de configuratie van de computer waarop het wordt uitgevoerd. U kunt bijvoorbeeld de procedure [1-3] volgen. U krijgt dan de volgende grafische interface te zien:

• de interface bevat twee tabbladen [1-2]: het ene, [Request], is voor het verzoek aan de dienst voor het genereren van willekeurige getallen, het andere, [Response], is voor de weergave van de ontvangen getallen;
• in [3] wordt aangegeven hoeveel verzoeken men aan de dienst wil doen;
• in [4] wordt het gewenste interval voor het genereren van getallen aangegeven;
• in [5] zal het aantal door de dienst teruggestuurde waarden een willekeurig getal zijn binnen het door de gebruiker vastgestelde interval [minCount, maxCount];
• in [6] zal de dienst, voordat hij zijn antwoord terugstuurt, zal de dienst delay milliseconden wachten, waarbij delay een willekeurig getal is binnen het door de gebruiker vastgestelde interval [minDelay, maxDelay];
• standaard zal de laag [swing] de synchrone interface van de service aanspreken. Om de asynchrone laag aan te spreken, vinkt de gebruiker [7] aan. In dat geval wordt de generatiedienst uitgevoerd in threads die gescheiden zijn van de event loop van de grafische interface. De Rx-bibliotheek beschikt over verschillende strategieën voor het genereren van deze threads. De gebruiker kan zijn strategie selecteren via [8];
• het genereren van getallen gebeurt met de knop [9];

• in [10], weergave van de resultaten. We zullen de structuur hiervan toelichten;
• in [11], het aantal verkregen resultaten;
• in [12], de uitvoeringstijd in milliseconden;
• in [13] heeft de gebruiker de mogelijkheid om de uitvoering te annuleren;

Elk resultaat heeft de volgende vorm:

{"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]: het nummer van de aanvraag. Er wordt aan herinnerd dat er meerdere aanvragen bij de generatieservice worden ingediend;
• [delay]: de wachttijd in milliseconden die de dienst heeft gemeten voordat het resultaat werd verzonden;
• [aleas]: de willekeurige getallen die door de dienst zijn teruggestuurd;
• [executedOn]: de naam van de thread waarin de service is uitgevoerd;
• [observedOn]: de naam van de thread die het resultaat heeft weergegeven. Bij een Swing-interface kan dit alleen de thread van de event loop zijn, in dit geval [AWT-EventQueue-0];
• [requestAt]: het tijdstip van de aanvraag in de vorm [heures:minutes:secondes:millisecondes];
• [responseAt]: het tijdstip waarop de resultaten zijn ontvangen, in dezelfde vorm;

We zullen nu de stukken code presenteren die nuttig zijn voor het begrijpen van het voorbeeld.

2.3. De synchrone interface

De servicelaag [1] heeft de volgende interface:

public interface IService {
  // willekeurige getallen in [a,b]
  // n getallen worden gegenereerd met n willekeurige getallen in het interval [minCount, maxCount]
  // de getallen worden gegenereerd na een wachttijd van delay milliseconden,
  // waarbij [delay] een willekeurig getal is in het interval [minDelay, maxDelay]
  public ServiceResponse getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay);
}

Het antwoord [ServiceResponse] is als volgt:

public class ServiceResponse {

  // wachttijd van de dienst
  private int delay;
  // willekeurige getallen
  private List<Integer> aleas;
  // uitvoeringsthread
  private String executedOn;

  // constructors

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

  // getters en setters
...
}

Het antwoord bestaat uit drie elementen:

• regel 6: de gegenereerde willekeurige getallen;
• regel 4: de wachttijd die de service in acht neemt voordat het resultaat wordt weergegeven;
• regel 8: de uitvoeringsthread van de dienst;

2.4. De synchrone aanroep

We gaan nu dieper in op de synchrone aanroep [4] die de laag [swing] doet naar de service [1]:

  private void doGenerateWithService() {
    // begin van de wachttijd
    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);
    }
    // einde wachtrij
    endWaiting();
}

• regels 5-12: de uitvoeringslus van de door de gebruiker aangevraagde [nbRequests]-verzoeken;
• regel 8: [service] is de implementatie van de synchrone interface [IService] die in paragraaf 2.3 is beschreven;
• regel 10: [model] is het model dat wordt weergegeven door de component JList van het tabblad [Response]. De elementen van dit sjabloon zijn de tekenreeksen jSON van de volgende elementen van het type [UiResponse]:

public class UiResponse {

  // klant-id
  private int idClient;
  // antwoord van de service
  private ServiceResponse serviceResponse;
  // naam van de observatiethread
  private String observedOn;
  // tijdstip van het verzoek
  private String requestAt;
  // tijdstip van het antwoord
  private String responseAt;

  // constructors

  public UiResponse() {
    observedOn = Thread.currentThread().getName();
    requestAt = getTimeStamp();
  }
  // privé-methoden

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

  // getters en setters
...
}

• regel 6: het antwoord van de dienst voor het genereren van getallen;
• regel 4: het nummer van de aanvraag waarop wordt geantwoord;
• regel 8: de thread voor de weergave van dit antwoord. Zoals gezegd zal dit altijd de thread van de event loop zijn;
• regels 10 en 12: het tijdstip van de aanvraag en dat van het antwoord;

2.5. Testen van synchrone aanroepen

We voeren de volgende configuratie uit:

We krijgen de volgende resultaten in het tabblad [Response]:

• in [1-2] hebben we inderdaad 10 antwoorden ontvangen, zoals gevraagd. Ze zijn op de eerste plaats ingevoegd in de volgorde waarin ze binnenkwamen. We zien dat ze in de volgorde van de verzoeken zijn ontvangen;
• ze zijn allemaal uitgevoerd en weergegeven in de thread van de event loop [AWT-EventQueue-0]. De verzoeken zijn dus achtereenvolgens in deze thread uitgevoerd. Er waren geen gelijktijdige verzoeken;
• wat hier niet zichtbaar is, is dat tijdens de uitvoering de grafische interface bevroren is. Het is bijvoorbeeld niet mogelijk om naar het tabblad [Response] te gaan om de binnenkomende antwoorden te bekijken of de uitvoering te onderbreken met de knop [Annuler]. Zelfs als deze knop op het tabblad [Request] had gestaan, zou hij onbruikbaar zijn geweest. Er zouden dan namelijk twee gebeurtenissen plaatsvinden:
  • het klikken op de knop [Générer];
  • het klikken op de knop [Annuler];

De klik op de knop [Annuler] wordt pas verwerkt nadat de bewerking die door de klik op de knop [Générer] is gestart, is voltooid. We hebben zojuist gezien dat deze bewerking de thread van de event loop gedurende de gehele uitvoering in beslag nam, waardoor de klik op de knop [Annuler] niet kon worden verwerkt. Dit is typisch het soort situaties waarin Rx een aanzienlijke verbetering kan bieden;

2.6. De asynchrone interface en de implementatie ervan

We richten ons nu op de interface van de laag [2] en de implementatie ervan met Rx. Deze zal niet meteen begrijpelijk zijn. We willen alleen de eenvoud van de code van deze implementatie benadrukken.

De asynchrone interface ziet er als volgt uit:

public interface IRxService {
  // willekeurige getallen in [a,b]
  // n getallen worden gegenereerd met n willekeurige getallen in het interval [minCount, maxCount]
  // de getallen worden gegenereerd na een wachttijd van delay milliseconden,
  // waarbij [delay] een willekeurig getal is in het interval [minDelay, maxDelay]
  public Observable<UiResponse> getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay, UiResponse uiResponse);
}

De verschillen met de synchrone interface die in paragraaf 2.3 is gepresenteerd, zijn als volgt:

• de klasse [UiResponse] die in paragraaf 2.3 wordt gepresenteerd, maakt nu deel uit van de parameters van de methode [getAleas] (regel 6). De reden hiervoor is dat, aangezien de verzoeken nu parallel worden uitgevoerd en de service een willekeurige tijd wacht voordat hij het resultaat retourneert, de antwoorden niet in de volgorde van de verzoeken bij ons terugkomen. We geven daarom het object [UiResponse] door, dat onder andere het nummer van het verzoek bevat:

  // klant-id (verzoek)
  private int idClient;
  // antwoord van de dienst
  private ServiceResponse serviceResponse;
  // naam van de observatiethread
  private String observedOn;
  // tijdstip van de aanvraag
  private String requestAt;
  // tijdstip van het antwoord
  private String responseAt;

• het type van het antwoord van de asynchrone service is een type [Observable<UiResponse>]. Het type [Observable<>] wordt geleverd door de Rx-bibliotheek. Het resultaat van het type [Observable<UiResponse>] geeft aan dat de methode [getAleas] een stroom van waarden van het type [UiResponse] levert, waarden die één voor één naar hun observer worden gepusht;

Laten we nu eens kijken naar de implementatie van deze interface:

public class RxService implements IRxService {

  // dienst
  private IService service;

  // fabrikant
  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();
      }
    });
  }
}

• regels 7-9: aan de constructor wordt een verwijzing naar de synchrone interface [IService] doorgegeven. Deze interface zorgt voor het genereren van de willekeurige getallen;
• de observable die door de methode [getAleas] wordt geretourneerd, wordt geconstrueerd door de statische methode [Observable.create]. Deze methode maakt het mogelijk om een asynchrone implementatie te bouwen op basis van een synchrone implementatie;
• regel 13: de parameter van de statische methode [Observable.create] is hier een lambda-functie die als parameter een type [Subscriber] ontvangt, ook hier weer een Rx-type. Een [Subscriber] is een object dat zich abonneert op een stroom van observables, d.w.z. een stroom van gegevens die asynchroon worden geleverd. Hier worden drie methoden van deze abonnee gebruikt:
  • [Subscriber.onNext] om gegevens door te geven (regel 16);
  • [Subscriber.onError] om een uitzondering door te geven (regel 18);
  • [Subscriber.onCompleted] om de abonnee te laten weten dat de gegevensstroom is beëindigd (regel 20);

Er kunnen meerdere abonnees zijn op één en dezelfde observable. Hier hebben we slechts één abonnee die zich abonneert op een stream met één enkel gegeven, namelijk dat wat in de regels 15-16 wordt geproduceerd. Het gegeven wordt geproduceerd door de synchrone implementatie van de service (regel 15) en aan de abonnee geleverd (regel 16).

Hoewel dit alles waarschijnlijk nog wat onduidelijk is, valt de extreme beknoptheid van deze asynchrone implementatie van de service zeker op.

2.7. De asynchrone aanroep

We gaan nu dieper in op de synchrone aanroep [5] die de laag [swing] doet naar de service [2]:

private void doGenerateWithRxService() {
        // begin wachttijd
        beginWaiting();
        // willekeurige getallen worden opgevraagd
        Observable<UiResponse> observables = Observable.empty();
        for (int i = 0; i < nbRequests; i++) {
            UiResponse uiResponse = new UiResponse();
            uiResponse.setIdClient(i);
            // planner
            int schedulerIndex = jComboBoxSchedulers.getSelectedIndex();
            switch (schedulerIndex) {
            case 0:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.io()));
                break;
...
            }
        }
...
    }

• regels 6-10: uitvoering van de door de gebruiker aangevraagde [nbRequests]-verzoeken;
• regels 7-8: voorbereiding van het object [UiResponse] dat nodig is voor de methode [getAleas] van de asynchrone service (regel 13). Dit houdt voornamelijk in dat het verzoeknummer [idClient] wordt opgeslagen;
• regel 13: de methode [getAleas] van de asynchrone service wordt aangeroepen. Deze retourneert een object [Observable<UiResponse>]. Deze aanroep roept de synchrone service nog niet aan. Laten we teruggaan naar de code van de asynchrone methode [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();
      }
    });
}

De code in de regels 4-11, die de synchrone service aanroept, wordt pas uitgevoerd wanneer zich een abonnee aanmeldt. Zolang er geen abonnees zijn, wordt deze code niet uitgevoerd.

Laten we teruggaan naar de code van de methode [doGenerateWithRxService]:

• regel 5: er wordt een lege observable aangemaakt (er wordt niets geobserveerd);
• regel 13: er wordt een observable aangemaakt waarvan de stream de samenvoeging is van de asynchrone [nbRequests]-streams die gekoppeld zijn aan de [nbRequests]-verzoeken. Dit wordt bereikt met de methode [Observable.mergeWith], waarmee twee asynchrone streams kunnen worden samengevoegd. In de Rx-terminologie wordt [mergeWith] een streamoperator genoemd. Het bijzondere aan deze operatoren is dat het resultaat van de bewerking meestal weer een [Observable] is. Uiteindelijk, na regel 17, verwijst de variabele [observables] naar één enkele stream die bestaat uit de asynchrone antwoorden [nbRequests] van de asynchrone service;
• regel 13: de samenvoegbewerking had als volgt geschreven kunnen worden:

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

maar we hebben geschreven:

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

We hebben hier de operator [subscribeOn] toegepast op de observable [rxService.getAleas]. Zoals zo vaak is het resultaat opnieuw een observable. Met de operator [subscribeOn] kan worden gespecificeerd dat de observable moet worden uitgevoerd in een thread die wordt geleverd door een [Scheduler]. Er zijn verschillende mogelijke [Scheduler]-operatoren, afgestemd op verschillende situaties. In de grafische interface hebben we er verschillende voorgesteld om de effecten van de verschillende operatoren te bekijken:

Dit levert de volgende code op:

    private void doGenerateWithRxService() {
        // begin wachtrij
        beginWaiting();
        // willekeurige getallen opvragen
        Observable<UiResponse> observables = Observable.empty();
        for (int i = 0; i < nbRequests; i++) {
            UiResponse uiResponse = new UiResponse();
            uiResponse.setIdClient(i);
            // planner
            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;
            }
        }
...
}

Laten we teruggaan naar de code in de regels 12-14. De scheduler [Schedulers.io()] wijst aan elke observable een nieuwe thread toe. Als we de code volgen:

• regel 5: we hebben een lege observable;
• regel 13, iteratie 1: observables is de lijst [observable0/thread0] (observable observable0 uitgevoerd op thread thread0);
• regel 13, iteratie 2: observables is de lijst [observable0/thread0, observable1/thread1];
• enz...

Uiteindelijk, na regel 28, hebben we een observable die het resultaat is van de samenvoeging van de observables [nbRequests] die op verschillende [nbRequests]-threads worden uitgevoerd. Niet alle schedulers werken op deze manier, zoals we tijdens de tests zullen zien.

Laten we verdergaan met het bestuderen van de aanroepcode van de asynchrone service:

private void doGenerateWithRxService() {
        // begin wachtrij
        beginWaiting();
        // willekeurige getallen worden opgevraagd
        Observable<UiResponse> observables = Observable.empty();
        for (int i = 0; i < nbRequests; i++) {
        ...
        }
        // waarnemer
        observables = observables.observeOn(SwingScheduler.getInstance());
        // deze observables worden uitgevoerd
        subscriptions.add(observables.subscribe(uiResponse -> {
            updateUi(uiResponse);
        } , th -> {
            System.out.println(th);
            doCancel();
        } , this::doCancel));
    }

• we hebben gezien dat wanneer we bij regel 10 aankomen, er één enkele observable is, een samenvoeging van [nbRequests] observables die al dan niet op [nbRequests] verschillende threads kunnen worden uitgevoerd, afhankelijk van de door de gebruiker gekozen scheduler;
• regel 10: met de operator [observeOn] kan worden gespecificeerd op welke thread de gegevens uit de observable moeten worden opgehaald, in dit geval de objecten van het type [UiResponse]. In een Swing-interface heb je geen keuze. Elke update van de interface moet plaatsvinden in de thread van de event loop. Hier worden de gegevens van de observable weergegeven in een Swing-component van het type JList. De thread [SwingScheduler.getInstance()] vertegenwoordigt de thread van de event loop. De klasse [SwingScheduler] is niet afkomstig uit de bibliotheek RxJava, maar uit de daarvan afgeleide bibliotheek RxSwing;
• wanneer we bij regel 12 aankomen, is de synchrone service nog steeds niet aangeroepen omdat de observable van regel 10 nog geen abonnee heeft. De regels 12-17 geven er een aan, dankzij de operator [subscribe]. De parameters van deze operator zijn hier drie lambda-functies:
  • de eerste, [uiResponse -> {updateUi(uiResponse);}], accepteert als parameter een van de objecten [UiResponse] die door de observable worden geproduceerd. Ter herinnering: we zullen hier [nbRequests] objecten van dit type hebben. De bijbehorende methode, hier updateUi, moet dit resultaat verwerken;
  • De tweede [th -> {System.out.println(th);doCancel();}] accepteert als parameter een type [Throwable], in dit geval een uitzondering die zich heeft voorgedaan tijdens de uitvoering van de observable. De bijbehorende methode moet deze informatie verwerken. Hier wordt deze informatie op de console weergegeven (regel 15) en wordt de uitvoering afgebroken, waardoor bepaalde elementen van de grafische interface worden bijgewerkt;
  • de derde [this::doCancel] wordt aangeroepen wanneer de observable aangeeft dat er geen gegevens meer te verzenden zijn. Hier is de observable de verzameling van [nbRequests]-observables. De resulterende observable geeft aan dat deze klaar is wanneer alle observables waaruit deze bestaat zelf hebben gemeld dat ze hun werk hebben voltooid. Dus wanneer deze derde lambda-functie wordt uitgevoerd, zijn alle gegevens ontvangen. De lokale methode [doCancel] werkt de grafische interface bij om aan te geven dat de uitvoering is voltooid;

De variabele [subscriptions] is als volgt gedefinieerd:

    // de abonnementen op de observables
protected List<Subscription> subscriptions = new ArrayList<Subscription>();

Het type [Subscription] vertegenwoordigt een abonnement, d.w.z. de koppeling tussen een abonnee [Subscriber] en datgene wat hij volgt [Observable]. We hebben hier een lijst met abonnementen gebruikt, hoewel er in dit voorbeeld slechts één is. De lokale methode [doCancel], die wordt uitgevoerd wanneer de observable aangeeft dat er geen gegevens meer te verzenden zijn, is als volgt:

    @Override
    protected void doCancel() {
        // einde wachten
        endWaiting();
        // in het geval van abonnementen
        if (jCheckBoxRxSwing.isSelected() && subscriptions != null) {
            subscriptions.forEach(Subscription::unsubscribe);
        }
}

• regel 7 beëindigt het abonnement van alle abonnees op de observable;

Uit deze beknopte uitleg kunnen we de volgende kernpunten onthouden:

• het type [Observable] verwijst naar een stroom van waarden, die één voor één naar abonnees of waarnemers worden gepusht;
• het type [Subscriber] verwijst naar een abonnee van het type [Observable];
• het type [Subscription] verwijst naar een abonnement, d.w.z. de koppeling tussen een [Subscriber] en een [Observable];
• het type [Observable] ondersteunt operatoren van het type [mergeWith, empty, subscribeOn, observeOn, ...], die voor het merendeel observabelen genereren. Deze operatoren dienen om de observabel te configureren vóór de uitvoering ervan:
  • wat men wil observeren;
  • de thread waarop de observable wordt uitgevoerd;
  • de thread waarop de abonnee de gegevens van de observable ontvangt;
• er worden twee soorten observables onderscheiden: de [froid / cold] en de [chaud / hot]. Een koude observable wordt bij elke nieuwe abonnee volledig uitgevoerd. Als elke uitvoering dezelfde gegevens oplevert, ontvangt elke nieuwe abonnee dezelfde gegevens als de vorige. Een ‘warme’ observable produceert doorgaans continu gegevens. Wanneer een abonnee zich abonneert, ontvangt hij de gegevens die vanaf het tijdstip van zijn abonnement zijn verzonden. Hij ontvangt geen gegevens die mogelijk eerder zijn verzonden. In ons voorbeeld is de observable ‘koud’: deze wordt bij elke nieuwe abonnee volledig opnieuw uitgevoerd. Wat wordt er in ons voorbeeld eigenlijk uitgevoerd? Om dat te weten te komen, moeten we teruggaan naar de definitie van de geobserveerde observable:

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

Bij elke nieuwe abonnee wordt de lambda-functie, een parameter van de methode [Observable.create] (regel 3), opnieuw uitgevoerd. Het zijn dus de regels 4-11 die voor elke nieuwe abonnee [subscriber] worden uitgevoerd;

2.8. Testen van asynchrone oproepen

We beginnen met het tonen van het effect van de verschillende aangeboden schedulers. Hiervoor gebruiken we de volgende parameters:

We stellen in [1-2] kleine waarden in, zodat er, als de verzoeken op dezelfde thread worden uitgevoerd, toch niet te lang hoeft te worden gewacht.

2.8.1. met de scheduler [Schedulers.io]

De volgende punten vallen op:

• de antwoorden worden in een andere volgorde ontvangen dan die van de verzoeken (zie idClient);
• elk verzoek is in een andere thread uitgevoerd;
• de grafische interface is deze keer niet meer vastgelegd:
  • men kan van het ene tabblad naar het andere schakelen;
  • men ziet de gegevens binnenkomen;
  • men heeft geen tijd om de knop [Annuler] te zien omdat de uitvoering te snel verloopt. We zullen deze in een andere test extra benadrukken;

2.8.2. met de planner [Schedulers.computation]

We kunnen de volgende punten opmerken:

• de antwoorden worden in een andere volgorde ontvangen dan die van de verzoeken (zie idClient);
• de verzoeken zijn uitgevoerd in 8 threads;
• thread nr. 3 is gebruikt voor de verzoeken 8 en 0;
• thread nr. 4 is gebruikt voor de verzoeken 9 en 1;
• de overige verzoeken hebben elk een andere thread gekregen;

De scheduler [Schedulers.computation] gebruikt evenveel threads als er cores op de gebruikte machine zijn. Deze informatie wordt verkregen via de uitdrukking [Runtime.getRuntime().availableProcessors()].

2.8.3. met de scheduler [Schedulers.newThread]

De werking is vergelijkbaar met die van de scheduler [Schedulers.io].

2.8.4. met de schedulers [Schedulers.trampoline, Schedulers.immediate]

De werking is synchroon. Alle verzoeken worden uitgevoerd op de thread van de event loop. Dit resultaat mag niet worden veralgemeend, maar het geeft aan dat in dit specifieke voorbeeld beide schedulers synchroon hebben gewerkt.

2.9. Grensgevallen

In dit voorbeeld gaan we werken met schedulers die een asynchrone werking mogelijk maken. Allereerst verhogen we het aantal verzoeken tot 100 met de scheduler [Schedulers.computation], die hier met 8 threads werkt. We krijgen het volgende resultaat:

• in [1] is de knop [Annuler] aanwezig en bruikbaar (asynchrone werking);

Laten we de uitvoering nu tot het einde doorlopen:

Uit [2] blijkt dat het uitvoeren van de 100 verzoeken ongeveer 4 seconden heeft geduurd (op 8 threads).

Laten we nu diezelfde 100 verzoeken uitvoeren met de scheduler [Schedulers.newThread], die elk verzoek op een aparte thread uitvoert:

In [1] zien we dat het uitvoeren van de 100 verzoeken (op 100 threads) een halve seconde heeft geduurd. Dit is dus aanzienlijk sneller dan met de scheduler [Schedulers.computation].

Laten we nu 800 verzoeken uitvoeren onder dezelfde omstandigheden, nog steeds met de scheduler [Schedulers.newThread]. We krijgen de volgende resultaten:

De 800 verzoeken worden in ongeveer 1 seconde uitgevoerd.

Wanneer we dit aantal verhogen (tot meer dan 2500 verzoeken op mijn computer – uitgevoerd in 1,5 s – dit aantal is uiteraard sterk afhankelijk van de werkomgeving op het moment van uitvoering), krijgen we uiteindelijk de volgende uitzondering:

Er is dus sprake van een stack overflow. Uit tests blijkt dat de werking van de scheduler [Schedulers.newThread] niet deterministisch is. Het kan voorkomen dat de bovenstaande uitzondering optreedt, je vervolgens nieuwe tests uitvoert, daarna terugkeert naar de configuratie die de uitzondering veroorzaakte en deze dan niet meer krijgt.

2.10. Conclusion

We hebben een voorbeeld gegeven van het gebruik van de Rx-bibliotheek. Laten we samenvatten wat we hebben geleerd:

We zijn uitgegaan van de volgende architectuur:

• in [4] voerde de laag [swing] synchrone aanroepen uit naar de laag [service];
• in [5], de laag [swing] deed asynchrone aanroepen naar de laag [rxService], die op haar beurt [6] synchroon aanriep naar de laag [service];

Het eerste wat ons opviel, was dat de Rx-bibliotheek het mogelijk maakte om eenvoudig de asynchrone interface [rxService] te creëren op basis van de synchrone interface [service] (zie paragraaf 2.4). Dit is een belangrijke les, omdat het betekent dat we een synchrone applicatie gemakkelijk kunnen omzetten naar een asynchrone applicatie.

In de laag [swing] zijn twee afzonderlijke methoden geschreven:

• de ene om synchrone aanroepen naar de service te doen (zie paragraaf 2.4);
• de andere om asynchrone aanroepen naar de service te doen (zie paragraaf 2.7);

Het schrijven van asynchrone aanroepen bleek aanzienlijk complexer dan dat van synchrone aanroepen. Niettemin zullen degenen die ervaring hebben met concurrente programmering waarbij meerdere threads moeten worden gesynchroniseerd, merken dat de Rx-oplossing eenvoudiger te schrijven is en alle lastige synchronisatie- en communicatieproblemen tussen threads voorkomt. Bij het schrijven hebben we de volgende belangrijke punten onderscheiden:

• het type [Observable] verwijst naar een stroom van gebeurtenissen (waarden) die asynchroon kunnen zijn (maar niet noodzakelijkerwijs) en die kunnen worden geobserveerd;
• het type [Subscriber] verwijst naar een abonnee op een type [Observable];
• het type [Subscription] verwijst naar een abonnement, d.w.z. de koppeling tussen een [Subscriber] en een [Observable];
• het type [Observable] ondersteunt operatoren van het type [mergeWith, empty, subscribeOn, observeOn, ...], die voor het merendeel observabelen genereren. Deze operatoren dienen om de observabel te configureren vóór de uitvoering ervan:
  • wat men wil observeren;
  • de thread waarop de observable wordt uitgevoerd;
  • de thread waarop de abonnee de gegevens van de observable ontvangt;
• er worden twee soorten observables onderscheiden: de [froid / cold] en de [chaud / hot]. Een koude observable wordt bij elke nieuwe abonnee volledig uitgevoerd. Als elke uitvoering dezelfde gegevens oplevert, ontvangt elke nieuwe abonnee dezelfde gegevens als de vorige. Een ‘warme’ observable produceert doorgaans continu gegevens. Wanneer een abonnee zich abonneert, ontvangt hij de gegevens die vanaf het moment van zijn abonnement zijn verzonden. Hij ontvangt geen gegevens die mogelijk eerder zijn verzonden. In ons voorbeeld is de observable ‘koud’: deze wordt bij elke nieuwe abonnee volledig opnieuw uitgevoerd.

Nu we een voorbeeld hebben gezien dat het nut van de Rx-bibliotheek heeft aangetoond, gaan we deze in meer detail bespreken.

De Rx-bibliotheek bevat talrijke methoden met generieke parameters in hun signatuur. We zullen deze signaturen kort herhalen (paragraaf 3). De parameters van deze methoden zijn meestal functionele interfaces (Java 8), d.w.z. interfaces met slechts één enkele methode. De daadwerkelijke parameters moeten dan instanties van deze interfaces zijn. Vóór Java 8 was het gebruikelijk om een interface te implementeren met een anonieme klasse. Met Java 8, en als de interface een functionele interface is, is het beknopter om deze te implementeren met een lambda-functie. We zullen deze dus bespreken (paragraaf 4). Zodra dit is gebeurd, zullen we de klasse [Stream] (paragraaf 5) bespreken, waarmee Java-collecties met lambda-functies kunnen worden verwerkt. Deze klasse is interessant omdat de klasse [Observable] van RxJava hieruit:

• bepaalde methoden;
• dezelfde manier om methoden aan elkaar te koppelen om één en dezelfde observable te verwerken;

Vervolgens zullen we de functionele interfaces presenteren die specifiek zijn voor de bibliotheek RxJava (paragraaf 6). Daarna gaan we verder met de belangrijkste elementen van de Rx-bibliotheek [Observable, Subscriber, Subscription, opérateurs] (paragraaf 7). De klasse [Observable] bevat tientallen operatoren die op hun beurt meerdere keren zijn overladen. Dit zorgt in eerste instantie voor een grote complexiteit, omdat deze operatoren en hun overladingen soms slechts in één detail van elkaar verschillen en het zonder ervaring moeilijk is om te weten welke operator je moet gebruiken. We zullen slechts een beperkt aantal operatoren behandelen en meestal zullen we hun overladingen negeren.

Het hele voorgaande deel zal worden uitgevoerd met de bibliotheek RxJava in eenvoudige console-toepassingen. Zodra de bibliotheek RxJava beschikbaar is, zullen we deze gebruiken in twee soorten grafische toepassingen:

• in paragraaf 8 komen we terug op de voorbeeld-Swing-toepassing om deze nader te beschrijven. We zullen dan de bibliotheek RxSwing gebruiken;
• in paragraaf 9 zullen we een Android-toepassing maken met de bibliotheek RxAndroid;

Als dit allemaal achter de rug is, beschikt de lezer over de middelen om op eigen kracht verder te gaan. Het zal waarschijnlijk even duren voordat hij de Rx-bibliotheek intuïtief kan gebruiken. Ik vond deze bibliotheek bijzonder interessant. Ik vond haar echter moeilijk te begrijpen en de leertijd was lang. Ik hoop dat dit document die leertijd voor de lezer zal verkorten. Het lijkt me de moeite waard.