2. Przykładowe wprowadzenie

Moje pierwsze zetknięcia z RxJava miały miejsce dzięki kursom i samouczkom znalezionym w Internecie. Poza tym, że teoria opierała się na pojęciach, do których nie byłem przyzwyczajony i które trudno mi było zrozumieć, przede wszystkim nie widziałem, do czego to mogłoby się przydać w prawdziwym życiu. Zaczniemy więc od przedstawienia przykładu (mam nadzieję, że prostego), w którym wykorzystanie RxJava naprawdę upraszcza pisanie kodu, a na tej podstawie postaramy się zidentyfikować najważniejsze elementy tej biblioteki.

Biblioteka RxJava opiera się na następującej koncepcji: strumień elementów typu T Observable<T> jest obserwowany przez jednego lub więcej subskrybentów (obserwatorów, konsumentów) Subscriber<T>. Biblioteka RxJava umożliwia uruchomienie strumienia Observable<T> w wątku T1, a jego obserwatora Subscriber<T> w wątku T2, bez koniecznościmusiał martwić się o zarządzanie cyklem życia tych wątków oraz o naturalnie trudne problemy, takie jak współdzielenie danych między wątkami i ich synchronizacja w celu wykonania zadania globalnego. Ułatwia to zatem programowanie asynchroniczne.

Strumień typu Observable<T> generuje elementy typu T, które można obserwować w miarę ich generowania. Jeśli obserwator i obiekt obserwowalny (termin ten jest używany potocznie w odniesieniu do typu Observable<T>) znajdują się w tym samym wątku, wówczas obiekt obserwowalny może wygenerować element (i+1) dopiero wtedy, gdy obserwator skonsumował element i. Niewiele jest przypadków, w których taka architektura ma sens. Jeśli obserwator i obserwowalny nie znajdują się w tym samym wątku, wówczas obserwowalny i jego obserwator zachowują się autonomicznie: obserwowalny generuje elementy we własnym tempie, a obserwator je pobiera we własnym tempie. Właśnie na tym polega zaleta tej biblioteki. Do tej pory mówiliśmy zawsze o jednym obserwatorze. W rzeczywistości obserwowalny może mieć dowolną liczbę obserwatorów.

2.1. Architektura przykładowej aplikacji

Przykładowa aplikacja ma następującą architekturę:

• w pliku [1] warstwa usługowa dostarcza listy liczb losowych. Warstwa ta jest wykonywana w tym samym wątku, co metoda [swing], która z niej korzysta. Dostarcza ona zatem swoje liczby w sposób synchroniczny;
• w metodzie [2] cienka warstwa adaptacyjna zaimplementowana za pomocą RxJava pozwala przedstawić warstwie [swing] asynchroniczną implementację tej samej usługi: może ona być wykonywana w innym wątku niż metoda [swing], która z niej korzysta;
• wywołanie [4] jest synchroniczne, podczas gdy wywołanie [5-6] jest asynchroniczne;

Chcemy tutaj pokazać, że biblioteka Rx pozwala w prosty sposób przekształcić interfejs synchroniczny w asynchroniczny. Dlaczego jest to przydatne? Zdarzenia w interfejsie Swing są przetwarzane w wątku zwanym potocznie pętlą zdarzeń (event loop). Zdarzenia są umieszczane w kolejce i przetwarzane jedno po drugim. Zdarzenie Ei+1 może zostać przetworzone dopiero wtedy, gdy poprzednie zdarzenie Ei zostało całkowicie przetworzone. Dlatego ważne jest, aby obsługa zdarzenia trwała jak najkrócej, tak aby interfejs graficzny pozostawał responsywny. Czasami obsługa zdarzenia może zająć dużo czasu. Dzieje się tak, gdy wymaga ona dostępu do sieci. Jeśli nie chcemy, aby interfejs graficzny zawieszał się w sposób nie do przyjęcia dla użytkownika, dostęp do sieci musi odbywać się w wątkach oddzielonych od pętli zdarzeń, aby ją odciążyć. W ten sposób wkraczamy w dziedzinę programowania współbieżnego (wiele wątków działa równolegle), słusznie uważaną za trudną. Biblioteka Rx oferuje proste i eleganckie rozwiązanie tego problemu.

Aby zasymulować długotrwałe operacje, usługa z przykładu generuje liczby losowe po upływie określonego czasu oczekiwania, co pozwala zaobserwować zachowanie interfejsu graficznego.

2.2. L'exécutable

Plik wykonywalny przykładowej aplikacji znajduje się w folderze [dvp/executables] z przykładami:

Istnieje kilka sposobów uruchomienia archiwum [swing-01] w zależności od konfiguracji komputera, na którym ma zostać uruchomione. Można na przykład postępować zgodnie z instrukcją zawartą w pliku [1-3]. W ten sposób otrzymujemy następujący interfejs graficzny:

• interfejs zawiera dwie zakładki [1-2]: jedną [Request] służącą do wysyłania zapytania do usługi generującej liczby losowe, a drugą [Response] służącą do wyświetlania otrzymanych liczb;
• w polu [3] należy podać liczbę żądań, które mają zostać wysłane do serwisu;
• w [4] określa się przedział [a,b], w którym mają być generowane liczby;
• w polu [5] liczba wartości zwracanych przez serwis będzie liczbą losową z przedziału [minCount, maxCount] ustalonego przez użytkownika;
• w [6], przed wysłaniem odpowiedzi, usługa odczeka delay milisekund, gdzie delay jest liczbą losową z przedziału [minDelay, maxDelay] określonego przez użytkownika;
• domyślnie warstwa [swing] będzie kierować się do synchronicznego interfejsu usługi. Aby skorzystać z warstwy asynchronicznej, użytkownik zaznaczy opcję [7]. W takim przypadku usługa generowania będzie działać w wątkach oddzielonych od pętli zdarzeń interfejsu graficznego. Biblioteka Rx oferuje różne strategie generowania tych wątków. Użytkownik może wybrać strategię w polu [8];
• generowanie liczb odbywa się za pomocą przycisku [9];

• w [10] – wyświetlanie wyników. Wyjaśnimy strukturę tych wyników;
• w [11] – liczba uzyskanych wyników;
• w [12] – czas wykonania w milisekundach;
• w [13] użytkownik ma możliwość anulowania wykonania;

Każdy wynik ma następującą postać:

{"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]: numer żądania. Przypominamy, że do serwisu generującego wysyłanych jest kilka żądań;
• [delay]: czas oczekiwania w milisekundach, jaki usługa odnotowała przed wysłaniem wyniku;
• [aleas]: liczby losowe zwrócone przez usługę;
• [executedOn]: nazwa wątku, w którym uruchomiono usługę;
• [observedOn]: nazwa wątku, który wyświetlił wynik. W przypadku interfejsu Swing może to być wyłącznie wątek pętli zdarzeń, w tym przypadku [AWT-EventQueue-0];
• [requestAt]: czas wysłania żądania w postaci [heures:minutes:secondes:millisecondes];
• [responseAt]: czas otrzymania wyników w tej samej postaci;

Przedstawimy teraz fragmenty kodu przydatne do zrozumienia tego przykładu.

2.3. Interfejs synchroniczny

Warstwa usługowa [1] posiada następujący interfejs:

public interface IService {
  // liczby losowe w [a,b]
  // n liczb jest generowanych przy użyciu n liczb losowych z przedziału [minCount, maxCount]
  // liczby są generowane po upływie opóźnienia wynoszącego delay milisekund,
  // gdzie [delay] jest liczbą losową z przedziału [minDelay, maxDelay]
  public ServiceResponse getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay);
}

Odpowiedź [ServiceResponse] wygląda następująco:

public class ServiceResponse {

  // czas oczekiwania na obsługę
  private int delay;
  // liczby losowe
  private List<Integer> aleas;
  // wątek wykonawczy
  private String executedOn;

  // konstruktory

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

  // metody pobierające i ustawiające
...
}

Odpowiedź składa się z trzech elementów:

• wiersz 6: wygenerowane liczby losowe;
• wiersz 4: czas oczekiwania, jaki usługa zachowuje przed zwróceniem wyniku;
• wiersz 8: wątek wykonawczy usługi;

2.4. Wywołanie synchroniczne

Przeanalizujemy teraz wywołanie synchroniczne [4], które warstwa [swing] kieruje do usługi [1]:

  private void doGenerateWithService() {
    // początek oczekiwania
    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);
    }
    // koniec oczekiwania
    endWaiting();
}

• wiersze 5–12: pętla wykonywania żądań [nbRequests] zgłoszonych przez użytkownika;
• wiersz 8: [service] stanowi implementację synchronicznego interfejsu [IService] przedstawionego w paragrafie 2.3;
• wiersz 10: [model] to szablon wyświetlany przez komponent JList w zakładce [Response]. Elementami tego szablonu są ciągi znaków jSON zawierające następujące elementy typu [UiResponse]:

public class UiResponse {

  // identyfikator klienta
  private int idClient;
  // odpowiedź serwisu
  private ServiceResponse serviceResponse;
  // nazwa wątku obserwacyjnego
  private String observedOn;
  // czas wysłania żądania
  private String requestAt;
  // czas odpowiedzi
  private String responseAt;

  // konstruktorzy

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

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

  // metody pobierające i ustawiające
...
}

• wiersz 6: odpowiedź serwisu generującego liczby;
• wiersz 4: numer zapytania, na które udzielono odpowiedzi;
• wiersz 8: wątek wyświetlania tej odpowiedzi. Jak już wspomniano, będzie to zawsze wątek pętli zdarzeń;
• wiersze 10 i 12: czas wysłania żądania oraz czas otrzymania odpowiedzi;

2.5. Testy wywołań synchronicznych

Uruchamiamy następującą konfigurację:

W zakładce [Response] otrzymujemy następujące wyniki:

• w [1-2] otrzymaliśmy rzeczywiście 10 odpowiedzi, zgodnie z żądaniem. Zostały one umieszczone na pierwszej pozycji w kolejności ich nadejścia. Widać, że zostały one uzyskane w kolejności zapytań;
• wszystkie zostały wykonane i wyświetlone w wątku pętli zdarzeń [AWT-EventQueue-0]. Zapytania zostały zatem wykonane jedno po drugim w tym wątku. Nie wystąpiły żadne równoczesne zapytania;
• nie widać tu jednak, że podczas wykonywania interfejs graficzny jest zawieszony. Nie ma na przykład możliwości przejścia do zakładki [Response], aby zobaczyć napływające odpowiedzi, ani przerwania wykonywania za pomocą przycisku [Annuler]. Nawet gdyby ten przycisk znajdował się na zakładce [Request], nie dałoby się z niego skorzystać. W rzeczywistości miałyby wtedy miejsce dwa zdarzenia:
  • kliknięcie przycisku [Générer];
  • kliknięcie przycisku [Annuler];

Kliknięcie przycisku [Annuler] jest obsługiwane dopiero po zakończeniu operacji uruchomionej przez kliknięcie przycisku [Générer]. Jak właśnie widzieliśmy, operacja ta zajmowała wątek pętli zdarzeń przez cały czas trwania wykonania, uniemożliwiając tym samym obsługę kliknięcia przycisku [Annuler]. Jest to typowy przykład sytuacji, w której Rx może przynieść znaczną poprawę;

2.6. Interfejs asynchroniczny i jego implementacja

Skupimy się teraz na interfejsie warstwy [2] oraz na jego implementacji przy użyciu Rx. Nie będzie ona od razu zrozumiała. Chcemy po prostu podkreślić prostotę kodu tej implementacji.

Interfejs asynchroniczny wygląda następująco:

public interface IRxService {
  // liczby losowe w [a,b]
  // n liczb jest generowanych przy użyciu n liczb losowych z przedziału [minCount, maxCount]
  // liczby są generowane po upływie opóźnienia wynoszącego delay milisekund,
  // gdzie [delay] jest liczbą losową z przedziału [minDelay, maxDelay]
  public Observable<UiResponse> getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay, UiResponse uiResponse);
}

Różnice w stosunku do interfejsu synchronicznego przedstawionego w punkcie 2.3 są następujące:

• klasa [UiResponse] przedstawiona w paragrafie 2.3 stanowi teraz część parametrów metody [getAleas] (wiersz 6). Wynika to z faktu, że ponieważ żądania są teraz wykonywane równolegle, a usługa czeka przez losowy czas przed zwróceniem wyniku, odpowiedzi nie będą do nas wracać w kolejności zgodnej z kolejnością żądań. Przekazujemy zatem obiekt [UiResponse], który zawiera między innymi numer żądania:

  // identyfikator klienta (żądanie)
  private int idClient;
  // odpowiedź serwisu
  private ServiceResponse serviceResponse;
  // nazwa wątku obserwacyjnego
  private String observedOn;
  // czas wysłania żądania
  private String requestAt;
  // czas odpowiedzi
  private String responseAt;

• Typ odpowiedzi usługi asynchronicznej to [Observable<UiResponse>]. Typ [Observable<>] jest dostarczany przez bibliotekę Rx. Wynik typu [Observable<UiResponse>] wskazuje, że metoda [getAleas] dostarcza strumień wartości typu [UiResponse], które są przekazywane (pushed) pojedynczo do ich obserwatora;

Przyjrzyjmy się teraz implementacji tego interfejsu:

public class RxService implements IRxService {

  // usługa
  private IService service;

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

• wiersze 7–9: do konstruktora przekazywane jest odwołanie do interfejsu synchronicznego [IService]. To właśnie on będzie odpowiedzialny za generowanie liczb losowych;
• obserwowalny zwracany przez metodę [getAleas] jest tworzony przez metodę statyczną [Observable.create]. To właśnie ta metoda pozwala na stworzenie implementacji asynchronicznej na podstawie implementacji synchronicznej;
• wiersz 13: parametrem metody statycznej [Observable.create] jest tutaj funkcja lambda, która jako parametr przyjmuje typ [Subscriber], czyli ponownie typ Rx. Obiekt typu [Subscriber] to obiekt, który subskrybuje strumień obserwowalnych, tj. strumień danych dostarczanych asynchronicznie. Wykorzystujemy tutaj trzy metody tego subskrybenta:
  • [Subscriber.onNext] do przesłania mu danych (wiersz 16);
  • [Subscriber.onError] do przekazania mu wyjątku (wiersz 18);
  • [Subscriber.onCompleted] w celu poinformowania subskrybenta, że strumień danych został zakończony (wiersz 20);

Ten sam obserwowalny może mieć wielu subskrybentów. W tym przypadku mamy tylko jednego subskrybenta, który subskrybuje strumień zawierający jedną wartość, generowaną w wierszach 15–16. Dane są generowane przez synchroniczną implementację usługi (wiersz 15) i przekazywane subskrybentowi (wiersz 16).

Chociaż wszystko to prawdopodobnie pozostaje niejasne, nie sposób nie zwrócić uwagi na niezwykłą zwięzłość tej asynchronicznej implementacji usługi.

2.7. Wywołanie asynchroniczne

Przeanalizujmy teraz synchroniczne wywołanie [5], które warstwa [swing] kieruje do usługi [2]:

private void doGenerateWithRxService() {
        // początek oczekiwania
        beginWaiting();
        // żądanie liczb losowych
        Observable<UiResponse> observables = Observable.empty();
        for (int i = 0; i < nbRequests; i++) {
            UiResponse uiResponse = new UiResponse();
            uiResponse.setIdClient(i);
            // harmonogram
            int schedulerIndex = jComboBoxSchedulers.getSelectedIndex();
            switch (schedulerIndex) {
            case 0:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.io()));
                break;
...
            }
        }
...
    }

• wiersze 6–10: wykonanie zapytań [nbRequests] zgłoszonych przez użytkownika;
• wiersze 7–8: przygotowanie obiektu [UiResponse], którego potrzebuje metoda [getAleas] usługi asynchronicznej (wiersz 13). Polega to głównie na zapisaniu numeru [idClient] żądania;
• wiersz 13: wywoływana jest metoda [getAleas] usługi asynchronicznej. Zwraca ona obiekt [Observable<UiResponse>]. To wywołanie nie uruchamia jeszcze usługi synchronicznej. Wróćmy do kodu asynchronicznej metody [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();
      }
    });
}

Kod w wierszach 4–11, który wywoła usługę synchroniczną, jest wykonywany dopiero wtedy, gdy zgłosi się subskrybent. Dopóki nie ma subskrybentów, kod ten nie jest wykonywany.

Wróćmy do kodu metody [doGenerateWithRxService]:

• wiersz 5: tworzymy pusty obiekt obserwowalny (nic nie jest obserwowane);
• wiersz 13: tworzymy obserwowalny, którego strumieniem będzie połączenie asynchronicznych strumieni [nbRequests] powiązanych z żądaniami [nbRequests]. Osiąga się to za pomocą metody [Observable.mergeWith], która pozwala połączyć dwa asynchroniczne strumienie. W terminologii Rx [mergeWith] nazywany jest operatorem strumienia. Cechą charakterystyczną tych operatorów jest to, że wynikiem operacji jest w większości przypadków ponownie [Observable]. Ostatecznie, po wierszu 17, zmienna [observables] oznacza pojedynczy strumień złożony z asynchronicznych odpowiedzi [nbRequests] generowanych przez usługę asynchroniczną;
• wiersz 13: operację scalania można by zapisać w następujący sposób:

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

ale zapisaliśmy:

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

W tym przypadku zastosowaliśmy operator [subscribeOn] na obserwowalnej zmiennej [rxService.getAleas]. Jak to często bywa, wynikiem jest ponownie obserwowalna zmienna. Operator [subscribeOn] pozwala określić, że obserwowalna musi być wykonywana w wątku dostarczonym przez [Scheduler]. Istnieje kilka możliwych operatorów [Scheduler] dostosowanych do różnych sytuacji. W interfejsie graficznym zaproponowaliśmy kilka z nich, aby zobaczyć, jak działają:

Daje to następujący kod:

    private void doGenerateWithRxService() {
        // początek oczekiwania
        beginWaiting();
        // żądanie liczb losowych
        Observable<UiResponse> observables = Observable.empty();
        for (int i = 0; i < nbRequests; i++) {
            UiResponse uiResponse = new UiResponse();
            uiResponse.setIdClient(i);
            // harmonogram
            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;
            }
        }
...
}

Wróćmy do kodu w wierszach 12–14. Harmonogram [Schedulers.io()] przypisuje każdy obserwowalny do nowego wątku. Jeśli prześledzimy kod:

• wiersz 5: mamy pustą obserwowalną;
• wiersz 13, iteracja 1: observables to lista [observable0/thread0] (obserwowalna observable0 uruchomiona w wątku thread0);
• wiersz 13, iteracja 2: observables to lista [observable0/thread0, observable1/thread1];
• itd...

Ostatecznie, po wierszu 28, otrzymujemy obserwowalną wartość będącą wynikiem połączenia obserwowalnych [nbRequests], które są wykonywane na różnych wątkach [nbRequests]. Nie wszystkie harmonogramy działają w ten sposób, co zobaczymy podczas testów.

Kontynuujmy analizę kodu wywołującego usługę asynchroniczną:

private void doGenerateWithRxService() {
        // początek oczekiwania
        beginWaiting();
        // żądanie liczb losowych
        Observable<UiResponse> observables = Observable.empty();
        for (int i = 0; i < nbRequests; i++) {
        ...
        }
        // obserwator
        observables = observables.observeOn(SwingScheduler.getInstance());
        // wykonuje się te obserwowalne
        subscriptions.add(observables.subscribe(uiResponse -> {
            updateUi(uiResponse);
        } , th -> {
            System.out.println(th);
            doCancel();
        } , this::doCancel));
    }

• zauważyliśmy, że po dotarciu do linii 10 mamy do czynienia z jednym obserwowalnym, będącym połączeniem obserwowalnych [nbRequests], które mogą, ale nie muszą, być wykonywane na różnych wątkach [nbRequests], w zależności od harmonogramu wybranego przez użytkownika;
• wiersz 10: operator [observeOn] pozwala określić, w którym wątku chcemy pobierać dane pochodzące z obserwowalnej, w tym przypadku obiekty typu [UiResponse]. W interfejsie Swing nie ma wyboru. Każda aktualizacja interfejsu musi odbywać się w wątku pętli zdarzeń. W tym przypadku dane z obserwowalnego obiektu zostaną wyświetlone w komponencie Swing typu JList. Wątek [SwingScheduler.getInstance()] stanowi wątek pętli zdarzeń. Klasa [SwingScheduler] nie pochodzi z biblioteki RxJava, lecz z jej pochodnej RxSwing;
• gdy dochodzimy do linii 12, usługa synchroniczna nadal nie została wywołana, ponieważ obserwowalna z linii 10 nie ma jeszcze subskrybenta. Linie 12–17 przypisują jej subskrybenta dzięki operatorowi [subscribe]. Parametrami tego operatora są tutaj trzy funkcje lambda:
  • pierwsza, [uiResponse -> {updateUi(uiResponse);}], przyjmuje jako parametr jeden z obiektów [UiResponse] wygenerowanych przez obserwowalną. Przypomnijmy, że w tym przypadku będziemy mieli [nbRequests] obiektów tego typu. Powiązana metoda, w tym przypadku updateUi, musi wykorzystać ten wynik;
  • Druga funkcja [th -> {System.out.println(th);doCancel();}] przyjmuje jako parametr typ [Throwable], czyli w tym przypadku wyjątek, który wystąpił podczas wykonywania obserwowalnej. Powiązana metoda musi wykorzystać tę informację. W tym przypadku wyświetlamy ją na konsoli (wiersz 15) i przerywamy wykonywanie, co spowoduje aktualizację niektórych elementów interfejsu graficznego;
  • trzecia funkcja [this::doCancel] jest wywoływana, gdy obserwowalna sygnalizuje, że nie ma już danych do przesłania. W tym przypadku obserwowalna stanowi zbiór obserwowalnych [nbRequests]. Obserwowalna wynikowa wskaże, że zakończyła działanie, gdy wszystkie obserwowalne składające się na nią same zasygnalizują, że zakończyły swoją pracę. Zatem gdy ta trzecia funkcja lambda zostanie wykonana, oznacza to, że otrzymano wszystkie dane. Lokalna metoda [doCancel] aktualizuje interfejs graficzny, aby odzwierciedlić fakt, że wykonanie zostało zakończone;

Zmienna [subscriptions] jest zdefiniowana w następujący sposób:

    // subskrypcje obserwowalnych
protected List<Subscription> subscriptions = new ArrayList<Subscription>();

Typ [Subscription] reprezentuje subskrypcję, tj. powiązanie między subskrybentem [Subscriber] a tym, co obserwuje [Observable]. Wykorzystaliśmy tutaj listę subskrypcji, chociaż w tym przykładzie występuje tylko jedna. Metoda lokalna [doCancel], uruchamiana w momencie, gdy obserwowalny sygnalizuje brak danych do przesłania, wygląda następująco:

    @Override
    protected void doCancel() {
        // koniec oczekiwania
        endWaiting();
        // w przypadku subskrypcji
        if (jCheckBoxRxSwing.isSelected() && subscriptions != null) {
            subscriptions.forEach(Subscription::unsubscribe);
        }
}

• w wierszu 7 odpisuje się wszystkich subskrybentów obserwowalnego obiektu;

Z tego zwięzłego wyjaśnienia można wyciągnąć następujące kluczowe wnioski:

• typ [Observable] oznacza strumień wartości, które są przesyłane pojedynczo do subskrybentów lub obserwatorów;
• typ [Subscriber] oznacza subskrybenta typu [Observable];
• typ [Subscription] oznacza subskrypcję, tj. powiązanie między [Subscriber] a [Observable];
• typ [Observable] dopuszcza operatory [mergeWith, empty, subscribeOn, observeOn, ...], z których większość generuje obserwowalne. Operatory te służą do konfiguracji obserwowalnego przed jego wykonaniem:
  • co ma być obserwowane;
  • wątek, na którym obserwowalna wartość jest wykonywana;
  • wątek, w którym subskrybent odbiera dane z obserwowalnego obiektu;
• rozróżnia się dwa typy obserwowalnych: [froid / cold] i [chaud / hot]. Obserwowalny typu „zimnego” jest w całości wykonywany przy każdym nowym subskrybencie. Jeśli każde wykonanie generuje te same dane, każdy nowy subskrybent otrzymuje te same dane, co poprzedni. Obserwowalna typu „hot” zazwyczaj generuje dane w sposób ciągły. Gdy subskrybent subskrybuje obserwowalną, otrzymuje dane wysłane od momentu subskrypcji. Nie otrzymuje danych, które mogły zostać wysłane wcześniej. W naszym przykładzie obserwowalna jest typu „cold”: jest w całości ponownie wykonywana przy każdym nowym subskrybencie. Co tak naprawdę jest wykonywane w naszym przykładzie? Aby to ustalić, należy powrócić do definicji obserwowanej obserwowalnej:

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

Przy każdym nowym subskrybencie funkcja lambda, będąca parametrem metody [Observable.create] (wiersz 3), jest ponownie wykonywana. Oznacza to, że dla każdego nowego subskrybenta [subscriber] wykonywane są wiersze 4–11;

2.8. Testy wywołań asynchronicznych

Zaczynamy od przedstawienia działania różnych proponowanych harmonogramów. W tym celu wykorzystujemy następujące parametry:

W [1-2] ustawiamy małe wartości, aby w przypadku wykonywania zapytań w tym samym wątku nie trzeba było czekać zbyt długo.

2.8.1. z harmonogramem [Schedulers.io]

Można zauważyć następujące kwestie:

• odpowiedzi otrzymujemy w kolejności innej niż kolejność zapytań (patrz idClient);
• każde żądanie zostało przetworzone w innym wątku;
• interfejs graficzny nie jest już tym razem zablokowany:
  • można przechodzić między zakładkami;
  • widzimy, jak napływają dane;
  • nie ma czasu, by dostrzec przycisk [Annuler], ponieważ wykonanie przebiega zbyt szybko. Podkreślimy to w innym teście;

2.8.2. z harmonogramem [Schedulers.computation]

Można zauważyć następujące kwestie:

• odpowiedzi otrzymujemy w kolejności innej niż kolejność zapytań (por. idClient);
• zapytania zostały wykonane w 8 wątkach;
• wątek nr 3 został wykorzystany do zapytań 8 i 0;
• wątek nr 4 został wykorzystany do zapytań nr 9 i 1;
• pozostałe zapytania miały po jednym innym wątku;

Harmonogram [Schedulers.computation] wykorzystuje tyle wątków, ile jest rdzeni w używanym komputerze. Informację tę uzyskuje się za pomocą wyrażenia [Runtime.getRuntime().availableProcessors()].

2.8.3. z harmonogramem [Schedulers.newThread]

Działa to analogicznie do harmonogramu [Schedulers.io].

2.8.4. z harmonogramami [Schedulers.trampoline, Schedulers.immediate]

Działają one synchronicznie. Wszystkie zapytania są wykonywane w wątku pętli zdarzeń. Nie należy uogólniać tego wyniku, ale należy po prostu stwierdzić, że w tym konkretnym przykładzie oba harmonogramy działały synchronicznie.

2.9. Przypadki graniczne

W tym przykładzie będziemy pracować z harmonogramami, które umożliwiają działanie asynchroniczne. Najpierw zwiększamy liczbę żądań do 100 przy użyciu harmonogramu [Schedulers.computation], który w tym przypadku działa z 8 wątkami. Otrzymujemy następujący wynik:

• w [1] przycisk [Annuler] jest widoczny i można z niego korzystać (działanie asynchroniczne);

Teraz pozwólmy, aby wykonanie przebiegło do końca:

W [2] widać, że wykonanie 100 zapytań zajęło około 4 sekund (na 8 wątkach).

Teraz wykonajmy te same 100 zapytań za pomocą harmonogramu [Schedulers.newThread], który wykonuje każde zapytanie w osobnym wątku:

W przypadku harmonogramu [1] widać, że wykonanie 100 zapytań (na 100 wątkach) zajęło pół sekundy. Jest to zatem znacznie szybsze niż w przypadku harmonogramu [Schedulers.computation].

Teraz wykonajmy 800 zapytań w tych samych warunkach, nadal korzystając z harmonogramu [Schedulers.newThread]. Otrzymujemy następujące wyniki:

800 zapytań zostało wykonanych w ciągu około 1 sekundy.

Gdy zwiększymy tę liczbę (ponad 2500 zapytań na moim komputerze – wykonanych w 1,5 s – liczba ta jest oczywiście w dużym stopniu zależna od środowiska pracy w momencie wykonywania), pojawia się następujący błąd:

Mamy więc do czynienia z przepełnieniem stosu. Testy pokazują, że działanie harmonogramu [Schedulers.newThread] nie jest deterministyczne. Może pojawić się powyższy wyjątek, można przeprowadzić nowe próby, a następnie powrócić do konfiguracji, która spowodowała wyjątek, i już go nie zaobserwować.

2.10. Conclusion

Przedstawiliśmy przykład wykorzystania biblioteki Rx. Podsumujmy to, czego się nauczyliśmy:

Wychodziliśmy z następującej architektury:

• w [4] warstwa [swing] wykonywała synchroniczne wywołania do warstwy [service];
• w warstwie [5], warstwa [swing] wykonywała asynchroniczne wywołania do warstwy [rxService], która z kolei synchronicznie wywoływała warstwę [service] za pośrednictwem warstwy [6];

Pierwszą rzeczą, jaką zauważyliśmy, było to, że biblioteka Rx umożliwiała łatwe utworzenie interfejsu asynchronicznego [rxService] na podstawie interfejsu synchronicznego [service] (por. punkt 2.4). Jest to ważna lekcja, ponieważ oznacza to, że można z łatwością przekształcić aplikację synchroniczną w aplikację asynchroniczną.

W warstwie [swing] napisano dwie odrębne metody:

• jedna do wykonywania synchronicznych wywołań usługi (patrz punkt 2.4);
• druga do wykonywania wywołań asynchronicznych (patrz punkt 2.7);

Pisanie wywołań asynchronicznych okazało się znacznie bardziej skomplikowane niż pisanie wywołań synchronicznych. Niemniej jednak osoby, które zajmowały się programowaniem współbieżnym z wieloma wątkami wymagającymi synchronizacji, przekonają się, że rozwiązanie Rx jest prostsze w implementacji i pozwala uniknąć wszystkich problemów związanych z synchronizacją oraz komunikacją między wątkami, które są trudnymi zagadnieniami. Podczas pisania tego tekstu wyróżniliśmy następujące istotne kwestie:

• typ [Observable] oznacza strumień zdarzeń (wartości), które mogą być (ale nie muszą) asynchroniczne i które można obserwować;
• typ [Subscriber] oznacza subskrybenta typu [Observable];
• typ [Subscription] oznacza subskrypcję, tj. powiązanie między [Subscriber] a [Observable];
• typ [Observable] dopuszcza operatory [mergeWith, empty, subscribeOn, observeOn, ...], z których większość generuje obserwowalne wielkości. Operatory te służą do konfiguracji obserwowalnej wielkości przed jej wykonaniem:
  • co ma być obserwowane;
  • wątek, na którym obserwowalna jest wykonywana;
  • wątek, w którym subskrybent odbiera dane z obserwowalnego obiektu;
• rozróżnia się dwa typy obserwowalnych: [froid / cold] i [chaud / hot]. Obserwowalny typu „zimnego” jest w całości wykonywany przy każdym nowym subskrybencie. Jeśli każde wykonanie generuje te same dane, każdy nowy subskrybent otrzymuje te same dane, co poprzedni. Obserwowalna typu „hot” zazwyczaj generuje dane w sposób ciągły. Gdy subskrybent subskrybuje, otrzymuje dane wysłane od momentu subskrypcji. Nie otrzymuje danych, które mogły zostać wysłane wcześniej. W naszym przykładzie obserwowalna jest typu „cold”: jest w całości ponownie wykonywana przy każdym nowym subskrybencie.

Teraz, gdy zapoznaliśmy się z przykładem ilustrującym zalety biblioteki Rx, omówimy ją bardziej szczegółowo.

Biblioteka Rx zawiera wiele metod, których sygnatury zawierają parametry generyczne. Krótko omówimy te sygnatury (akapit 3). Parametry tych metod to w większości interfejsy funkcjonalne (Java 8), czyli interfejsy posiadające tylko jedną metodę. Rzeczywiste parametry muszą zatem być instancjami tych interfejsów. Przed wprowadzeniem Javy 8 zwyczajowo implementowano interfejs za pomocą klasy anonimowej. W Javie 8, jeśli interfejs jest interfejsem funkcjonalnym, bardziej zwięzłe jest implementowanie go za pomocą funkcji lambda. Omówimy więc te funkcje (akapit 4). Gdy to zrobimy, przedstawimy klasę [Stream] (akapit 5), która umożliwia przetwarzanie kolekcji Java za pomocą funkcji lambda. Klasa ta jest interesująca, ponieważ klasa [Observable] z pakietu RxJava zapożycza od niej:

• niektóre metody;
• ten sam sposób łączenia metod w łańcuch w celu przetwarzania tej samej obserwowalnej wartości;

Następnie przedstawimy interfejsy funkcjonalne specyficzne dla biblioteki RxJava (akapit 6). Kontynuujemy omówieniem głównych elementów biblioteki Rx [Observable, Subscriber, Subscription, opérateurs] (akapit 7). Klasa [Observable] zawiera kilkadziesiąt operatorów, które same w sobie są wielokrotnie przeciążone. Początkowo powoduje to dużą złożoność, ponieważ operatorzy ci i ich przeciążenia różnią się czasami jedynie jednym szczegółem i bez doświadczenia trudno jest określić, którego operatora należy użyć. Przedstawimy jedynie ograniczoną liczbę operatorów, a w większości przypadków pominiemy ich nadpisania.

Cała poprzednia część zostanie zrealizowana przy użyciu biblioteki RxJava w prostych aplikacjach konsolowych. Po opanowaniu biblioteki RxJava wykorzystamy ją w dwóch rodzajach aplikacji graficznych:

• w rozdziale 8 powrócimy do przykładowej aplikacji Swing, aby omówić ją bardziej szczegółowo. Wykorzystamy wówczas bibliotekę RxSwing;
• w punkcie 9 stworzymy aplikację na Androida z wykorzystaniem biblioteki RxAndroid;

Gdy wszystko to zostanie zrobione, czytelnik będzie miał narzędzia, by działać samodzielnie. Prawdopodobnie minie trochę czasu, zanim nauczy się intuicyjnie korzystać z biblioteki Rx. Uważam tę bibliotekę za szczególnie interesującą. Jednak jej zrozumienie okazało się dla mnie skomplikowane, a nauka jej obsługi zajęła mi sporo czasu. Mam nadzieję, że niniejszy dokument skróci ten czas dla czytelnika. Wydaje mi się, że gra jest warta świeczki.