2. Приклад вступу

Мої перші знайомства з RxJava відбулися завдяки курсам та підручникам, знайденим в Інтернеті. Окрім того, що в теорії використовувалися поняття, до яких я не був звиклий і які мені було важко зрозуміти, я, головне, не бачив, для чого це може знадобитися в реальному житті. Тому ми почнемо з наведення прикладу (сподіваюся, простого), де використання RxJava суттєво спрощує написання коду, і на цій основі спробуємо визначити найважливіші елементи цієї бібліотеки.

Бібліотека RxJava ґрунтується на такій концепції: потік елементів типу T Observable<T> спостерігається одним або кількома підписниками (абонентами, спостерігачами, споживачами) Subscriber<T>. Бібліотека RxJava дозволяє потоку Observable<T> виконуватися в потоці T1, а його спостерігачеві Subscriber<T> — у потоці T2, при цьому розробнику нетурбуватися про управління життєвим циклом цих потоків та про природно складні проблеми, такі як обмін даними між потоками та їх синхронізація для виконання загального завдання. Таким чином, вона спрощує асинхронне програмування.

Потік Observable<T> генерує елементи типу T, які можна спостерігати по мірі їх генерації. Якщо спостерігач та об’єкт спостереження (термін «об’єкт спостереження» вживається неточно для позначення типу Observable<T>) знаходяться в одному потоці, то об’єкт спостереження може генерувати елемент (i+1) лише тоді, коли спостерігач спожив елемент i. Випадків, коли така архітектура є доцільною, небагато. Якщо спостерігач і об’єкт спостереження не знаходяться в одному потоці, то об’єкт спостереження та його спостерігач поводяться автономно: об’єкт спостереження генерує елементи у своєму темпі, а спостерігач споживає їх у своєму темпі. Саме в цьому полягає цінність бібліотеки. Досі ми завжди говорили про одного спостерігача. Насправді об’єкт спостереження може мати довільну кількість спостерігачів.

2.1. Архітектура прикладу додатка

Приклад додатка має таку архітектуру:

• у [1] сервісний рівень видає списки випадкових чисел. Цей рівень виконується в тому самому потоці, що й метод [swing], який його використовує. Він видає свої числа синхронно;
• у [2] тонкий адаптаційний рівень, реалізований за допомогою RxJava, дозволяє надати рівню [swing] асинхронну реалізацію того самого сервісу: вона може виконуватися в іншому потоці, ніж метод [swing], який її використовує;
• виклик [4] є синхронним, тоді як виклик [5-6] — асинхронним;

Тут ми хочемо показати, що бібліотека Rx дозволяє легко перетворити синхронний інтерфейс на асинхронний. Чому це корисно? Події інтерфейсу Swing обробляються у потоці, який зазвичай називають циклом подій (event loop). Події ставляться в чергу та обробляються одна за одною. Подія Ei+1 може бути оброблена лише тоді, коли попередня подія Ei була повністю оброблена. Тому важливо, щоб обробка події була якомога коротшою, щоб графічний інтерфейс залишався чутливим. Іноді обробка події може займати багато часу. Це відбувається, якщо така обробка передбачає мережеві операції. Якщо ми не хочемо, щоб графічний інтерфейс зависав у спосіб, неприйнятний для користувача, то ці мережеві операції повинні виконуватися в потоках, відокремлених від циклу подій, щоб звільнити його. Тут ми вступаємо в сферу паралельного програмування (кілька потоків виконуються паралельно), яке справедливо вважається складним. Бібліотека Rx пропонує просте й елегантне рішення цієї проблеми.

Щоб імітувати тривалі обчислення, сервіс із прикладу видає випадкові числа після певної затримки, щоб можна було спостерігати за поведінкою графічного інтерфейсу.

2.2. L'exécutable

Виконуваний файл прикладу програми знаходиться у папці [dvp/executables] з прикладами:

Існують різні способи запуску архіву [swing-01] залежно від конфігурації комп’ютера, на якому його запускають. Наприклад, можна скористатися процесом [1-3]. У результаті з’явиться такий графічний інтерфейс:

• інтерфейс містить дві вкладки [1-2]: одну — [Request] — для запиту до служби генерації випадкових чисел, іншу — [Response] — для відображення отриманих чисел;
• у [3] вказується, скільки запитів потрібно надіслати до сервісу;
• у [4] вказується інтервал [a,b], у якому мають генеруватися бажані числа;
• у полі [5] кількість значень, що повертаються сервісом, буде випадковим числом у заданому користувачем інтервалі [minCount, maxCount];
• у [6], перед тим як надіслати відповідь, сервіс зачекає delay мілісекунд, де delay — це випадкове число в інтервалі [minDelay, maxDelay], заданому користувачем;
• за замовчуванням рівень [swing] звертатиметься до синхронного інтерфейсу служби. Щоб звернутися до асинхронного рівня, користувач має встановити прапорець [7]. У цьому випадку служба генерації виконуватиметься в потоках, відокремлених від циклу подій графічного інтерфейсу. Бібліотека Rx має різні стратегії створення цих потоків. Користувач зможе вибрати свою стратегію в [8];
• генерація чисел здійснюється за допомогою кнопки [9];

• у [10] — відображення результатів. Ми пояснимо їхню структуру;
• у [11] — кількість отриманих результатів;
• у [12] — час виконання у мілісекундах;
• у [13] користувач має можливість скасувати виконання;

Кожен результат має такий вигляд:

{"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]: номер запиту. Нагадуємо, що до служби генерації надсилається кілька запитів;
• [delay]: час очікування в мілісекундах, який сервіс зафіксував перед надсиланням результату;
• [aleas]: випадкові числа, повернуті службою;
• [executedOn]: ім’я потоку, у якому виконувалася служба;
• [observedOn]: ім’я потоку, який вивів результат. У разі інтерфейсу Swing це може бути лише потік циклу подій, у даному випадку [AWT-EventQueue-0];
• [requestAt]: час запиту у форматі [heures:minutes:secondes:millisecondes];
• [responseAt]: час отримання результатів у тому ж форматі;

Тепер ми розглянемо фрагменти коду, необхідні для розуміння прикладу.

2.3. Синхронний інтерфейс

Рівень сервісу [1] має такий інтерфейс:

public interface IService {
  // випадкові числа в [a,b]
  // n чисел генерується з n випадкових чисел в інтервалі [minCount, maxCount]
  // числа генеруються після затримки в delay мілісекунд,
  // де [delay] — випадкове число в інтервалі [minDelay, maxDelay]
  public ServiceResponse getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay);
}

Відповідь [ServiceResponse] виглядає так:

public class ServiceResponse {

  // час очікування служби
  private int delay;
  // випадкові числа
  private List<Integer> aleas;
  // потік виконання
  private String executedOn;

  // конструктори

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

  // методи getter та setter
...
}

Відповідь складається з трьох елементів:

• рядок 6: згенеровані випадкові числа;
• рядок 4: час очікування, який дотримує служба перед поверненням результату;
• рядок 8: потік виконання сервісу;

2.4. Синхронний виклик

Тепер детально розглянемо синхронний виклик [4], який рівень [swing] здійснює до служби [1]:

  private void doGenerateWithService() {
    // початок очікування
    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);
    }
    // кінець очікування
    endWaiting();
}

• рядки 5–12: цикл виконання запитів [nbRequests], що надходять від користувача;
• рядок 8: [service] — це реалізація синхронного інтерфейсу [IService], представленого в розділі 2.3;
• рядок 10: [model] — це модель, що відображається компонентом JList на вкладці [Response]. Елементами цієї моделі є рядки jSON, що містять елементи типу [UiResponse], а саме:

public class UiResponse {

  // ідентифікатор клієнта
  private int idClient;
  // відповідь сервісу
  private ServiceResponse serviceResponse;
  // ім'я потоку спостереження
  private String observedOn;
  // час запиту
  private String requestAt;
  // час відповіді
  private String responseAt;

  // конструктори

  public UiResponse() {
    observedOn = Thread.currentThread().getName();
    requestAt = getTimeStamp();
  }
  // приватні методи

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

  // гетери та сеттери
...
}

• рядок 6: відповідь служби генерації чисел;
• рядок 4: номер запиту, на який надано відповідь;
• рядок 8: потік відображення цієї відповіді. Як уже зазначалося, це завжди буде потік циклу подій;
• рядки 10 і 12: час запиту та час відповіді;

2.5. Тестування синхронних викликів

Ми виконуємо таку конфігурацію:

У вкладці [Response] ми отримуємо такі результати:

• у [1-2] ми дійсно отримали 10 відповідей, як і було задано. Вони були вставлені на перше місце у порядку їх надходження. Бачимо, що вони були отримані у порядку запитів;
• усі вони були виконані та відображені у потоці циклу подій [AWT-EventQueue-0]. Отже, запити виконувалися один за одним у цьому потоці. Одночасних запитів не було;
• що тут не видно, так це те, що під час виконання графічний інтерфейс зависає. Наприклад, немає можливості перейти на вкладку [Response], щоб побачити, як надходять відповіді, або перервати виконання за допомогою кнопки [Annuler]. Навіть якби ця кнопка була на вкладці [Request], нею не можна було б скористатися. Адже тоді відбувалися б дві події:
  • натискання кнопки [Générer];
  • натискання на кнопку [Annuler];

Клік на кнопку [Annuler] обробляється лише після завершення операції, запущеної кліком на кнопку [Générer]. Ми щойно побачили, що ця операція займала потік циклу подій протягом усього часу виконання, тим самим перешкоджаючи обробці натискання кнопки [Annuler]. Це типовий приклад ситуації, в якій Rx може забезпечити суттєве поліпшення;

2.6. Асинхронний інтерфейс та його реалізація

Тепер ми розглянемо інтерфейс рівня [2], а також його реалізацію за допомогою Rx. Вона не буде зрозумілою відразу. Ми просто хочемо підкреслити простоту коду цієї реалізації.

Асинхронний інтерфейс має такий вигляд:

public interface IRxService {
  // випадкові числа в [a,b]
  // n чисел генерується з n випадкових чисел в інтервалі [minCount, maxCount]
  // числа генеруються після очікування тривалістю delay мілісекунд,
  // де [delay] — випадкове число в інтервалі [minDelay, maxDelay]
  public Observable<UiResponse> getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay, UiResponse uiResponse);
}

Відмінності від синхронного інтерфейсу, представленого в розділі 2.3, такі:

• клас [UiResponse], представлений у розділі 2.3, тепер є частиною параметрів методу [getAleas] (рядок 6). Причина полягає в тому, що, оскільки запити тепер виконуються паралельно, а служба очікує випадковий проміжок часу перед поверненням результату, відповіді не надходитимуть до нас у порядку надходження запитів. Тому ми передаємо об’єкт [UiResponse], який, серед іншого, містить номер запиту:

  // ідентифікатор клієнта (запит)
  private int idClient;
  // відповідь сервісу
  private ServiceResponse serviceResponse;
  //— ім’я потоку спостереження
  private String observedOn;
  // час запиту
  private String requestAt;
  // час відповіді
  private String responseAt;

• тип відповіді асинхронного сервісу — це тип [Observable<UiResponse>]. Тип [Observable<>] надається бібліотекою Rx. Результат типу [Observable<UiResponse>] вказує, що метод [getAleas] надає потік значень типу [UiResponse], які по черзі передаються (pushed) до їхнього спостерігача;

Тепер розглянемо реалізацію цього інтерфейсу:

public class RxService implements IRxService {

  // сервіс
  private IService service;

  // виробник
  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();
      }
    });
  }
}

• рядки 7–9: конструктору передається посилання на синхронний інтерфейс [IService]. Саме він буде виконувати генерацію випадкових чисел;
• спостерігабельний об’єкт, що повертається методом [getAleas], створюється статичним методом [Observable.create]. Саме цей метод дозволяє побудувати асинхронну реалізацію на основі синхронної;
• рядок 13: параметром статичного методу [Observable.create] тут є лямбда-функція, яка приймає як параметр тип [Subscriber], що, знову ж таки, є типом Rx. [Subscriber] — це об’єкт, який підписується на потік спостережуваних величин, тобто потік даних, що передаються асинхронно. Тут використовуються три методи цього підписника:
  • [Subscriber.onNext] — для передачі йому даних (рядок 16);
  • [Subscriber.onError] — для передачі йому винятку (рядок 18);
  • [Subscriber.onCompleted] — для повідомлення підписника про завершення потоку даних (рядок 20);

Один об’єкт спостереження може мати декількох підписників. У цьому випадку ми матимемо лише одного підписника, який підписується на потік з одним єдиним значенням, що генерується у рядках 15–16. Значення генерується синхронною реалізацією сервісу (рядок 15) і передається підписнику (рядок 16).

Хоча все це, ймовірно, залишається незрозумілим, не можна не відзначити надзвичайну лаконічність цієї асинхронної реалізації сервісу.

2.7. Асинхронний виклик

Тепер детально розглянемо синхронний виклик [5], який рівень [swing] здійснює до сервісу [2]:

private void doGenerateWithRxService() {
        // початок очікування
        beginWaiting();
        // запит на випадкові числа
        Observable<UiResponse> observables = Observable.empty();
        for (int i = 0; i < nbRequests; i++) {
            UiResponse uiResponse = new UiResponse();
            uiResponse.setIdClient(i);
            // планувальник
            int schedulerIndex = jComboBoxSchedulers.getSelectedIndex();
            switch (schedulerIndex) {
            case 0:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.io()));
                break;
...
            }
        }
...
    }

• рядки 6–10: виконання запитів [nbRequests], які надсилає користувач;
• рядки 7–8: підготовка об’єкта [UiResponse], необхідного для методу [getAleas] асинхронного сервісу (рядок 13). В основному це полягає у записі номера запиту [idClient];
• рядок 13: викликається метод [getAleas] асинхронної служби. Він повертає об’єкт [Observable<UiResponse>]. Цей виклик ще не запускає синхронну службу. Повернемося до коду асинхронного методу [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();
      }
    });
}

Код у рядках 4–11, який викликає синхронну службу, виконується лише тоді, коли з’являється підписник. Доки підписників немає, цей код не виконується.

Повернемося до коду методу [doGenerateWithRxService]:

• рядок 5: створюється порожній об’єкт-спостерігач (нічого не спостерігається);
• рядок 13: створюється об’єкт, потік якого буде об’єднанням асинхронних потоків [nbRequests], пов’язаних із запитами [nbRequests]. Це досягається за допомогою методу [Observable.mergeWith], який дозволяє об’єднати два асинхронні потоки. У термінології Rx [mergeWith] називають оператором потоку. Особливістю цих операторів є те, що результатом операції здебільшого знову є [Observable]. У підсумку, після рядка 17 змінна [observables] позначає єдиний потік, що складається з асинхронних відповідей [nbRequests], наданих асинхронним сервісом;
• рядок 13: операцію об’єднання можна було б записати так:

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

але ми написали:

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

Тут ми використали оператор [subscribeOn] для спостережуваної величини [rxService.getAleas]. Як це часто буває, результатом знову є спостережувана величина. Оператор [subscribeOn] дозволяє вказати, що спостережувана величина має виконуватися у потоці, наданому [Scheduler]. Існує кілька можливих варіантів [Scheduler], пристосованих до різних ситуацій. У графічному інтерфейсі ми запропонували кілька з них, щоб побачити, як вони впливають один на одного:

Це дає такий код:

    private void doGenerateWithRxService() {
        // початок очікування
        beginWaiting();
        // запит на випадкові числа
        Observable<UiResponse> observables = Observable.empty();
        for (int i = 0; i < nbRequests; i++) {
            UiResponse uiResponse = new UiResponse();
            uiResponse.setIdClient(i);
            // планувальник
            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;
            }
        }
...
}

Повернемося до коду рядків 12–14. Планувальник [Schedulers.io()] призначає кожному об’єкту спостереження новий потік. Якщо простежити код:

• рядок 5: маємо порожній об’єкт-спостерігач;
• рядок 13, ітерація 1: observables — це список [observable0/thread0] (спостережуваний об’єкт observable0, що виконується у потоці thread0);
• рядок 13, ітерація 2: observables — це список [observable0/thread0, observable1/thread1];
• тощо...

У підсумку, після рядка 28 ми отримуємо об’єкт-спостерігач, що є результатом злиття об’єктів-спостерігачів [nbRequests], які виконуються на різних потоках [nbRequests]. Не всі планувальники працюють саме так, як ми побачимо під час тестування.

Продовжимо розгляд коду виклику асинхронної служби:

private void doGenerateWithRxService() {
        // початок очікування
        beginWaiting();
        // запит на випадкові числа
        Observable<UiResponse> observables = Observable.empty();
        for (int i = 0; i < nbRequests; i++) {
        ...
        }
        // спостерігач
        observables = observables.observeOn(SwingScheduler.getInstance());
        // виконуються ці спостережувані величини
        subscriptions.add(observables.subscribe(uiResponse -> {
            updateUi(uiResponse);
        } , th -> {
            System.out.println(th);
            doCancel();
        } , this::doCancel));
    }

• ми бачили, що, коли доходимо до рядка 10, маємо єдиний об’єкт спостереження, який є злиттям [nbRequests] об’єктів спостереження, що можуть виконуватися на [nbRequests] різних потоках або ні, залежно від планувальника, обраного користувачем;
• рядок 10: оператор [observeOn] дозволяє вказати, у якому потоці потрібно отримувати дані з об’єкта спостереження, в даному випадку об’єкти типу [UiResponse]. У інтерфейсі Swing вибору немає. Будь-яке оновлення інтерфейсу має відбуватися у потоці циклу подій. У цьому випадку дані з об’єкта спостереження будуть відображатися у компоненті Swing JList. Пот [SwingScheduler.getInstance()] є потоком циклу подій. Клас [SwingScheduler] походить не з бібліотеки RxJava, а з похідної бібліотеки RxSwing;
• коли доходимо до рядка 12, синхронна служба все ще не була викликана, оскільки об’єкт спостереження з рядка 10 ще не має підписника. Рядки 12–17 надають йому підписника завдяки оператору [subscribe]. Параметрами цього оператора тут є три лямбда-функції:
  • перша, [uiResponse -> {updateUi(uiResponse);}], приймає як параметр один із об’єктів [UiResponse], створених спостережуваним об’єктом. Нагадаємо, що тут ми матимемо [nbRequests] об’єктів цього типу. Відповідний метод, у даному випадку updateUi, повинен використовувати цей результат;
  • друга функція [th -> {System.out.println(th);doCancel();}] приймає як параметр об’єкт типу [Throwable] — у даному випадку це виняток, що виник під час виконання спостережуваного об’єкта. Відповідний метод повинен використовувати цю інформацію. Тут ми виводимо її на консоль (рядок 15) і припиняємо виконання, що призведе до оновлення деяких елементів графічного інтерфейсу;
  • третій [this::doCancel] викликається, коли обсервабел повідомляє, що у нього більше немає даних для передачі. Тут обсервабел є об’єднанням обсервабелів [nbRequests]. Результуюча спостережувана вкаже, що вона завершила роботу, коли всі спостережувані, з яких вона складається, самі повідомлять про завершення своєї роботи. Отже, коли ця третя лямбда-функція виконується, ми вже отримали всі дані. Локальний метод [doCancel] оновлює графічний інтерфейс, щоб відобразити завершення виконання;

Змінна [subscriptions] визначається наступним чином:

    // підписки на спостережувані величини
protected List<Subscription> subscriptions = new ArrayList<Subscription>();

Тип [Subscription] представляє підписку, тобто зв’язок між підписником [Subscriber] та тим, що він спостерігає ([Observable]). Тут ми використали список підписок, хоча в цьому прикладі є лише одна. Локальний метод [doCancel], що виконується, коли об’єкт спостереження повідомляє, що у нього більше немає даних для передачі, виглядає так:

    @Override
    protected void doCancel() {
        // кінець очікування
        endWaiting();
        // у разі підписок
        if (jCheckBoxRxSwing.isSelected() && subscriptions != null) {
            subscriptions.forEach(Subscription::unsubscribe);
        }
}

• у рядку 7 скасовується підписка всіх підписників на об’єкт спостереження;

З цього короткого пояснення можна виокремити такі ключові моменти:

• тип [Observable] позначає потік значень, які по черзі надсилаються підписникам або спостерігачам;
• тип [Subscriber] позначає підписника типу [Observable];
• тип [Subscription] позначає підписку, тобто зв’язок між [Subscriber] та [Observable];
• тип [Observable] допускає оператори [mergeWith, empty, subscribeOn, observeOn, ...], які здебільшого генерують спостережувані величини. Ці оператори слугують для налаштування спостережуваної величини перед її виконанням:
  • що саме потрібно спостерігати;
  • потік, на якому виконується об’єкт спостереження;
  • потік, у якому підписник отримує дані від об’єкта спостереження;
• розрізняють два типи об’єктів спостереження: [froid / cold] та [chaud / hot]. «Холодний» об’єкт спостереження повністю виконується при підключенні кожного нового підписника. Якщо кожне виконання генерує однакові дані, кожен новий підписник отримує ті самі дані, що й попередній. «Гарячий» обсервабель зазвичай генерує дані безперервно. Коли підписник підписується, він отримує дані, що були відправлені з моменту його підписки. Він не отримує даних, які могли бути відправлені раніше. У нашому прикладі обсервабел є «холодним»: він повністю перевиконується при кожному новому підписнику. Що насправді виконується в нашому прикладі? Щоб це з’ясувати, потрібно повернутися до визначення обсервабела, що спостерігається:

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

З кожним новим підписником функція лямбда, що є параметром методу [Observable.create] (рядок 3), виконується заново. Отже, для кожного нового підписника [subscriber] виконуються рядки 4–11;

2.8. Тестування асинхронних викликів

Спочатку ми продемонструємо вплив різних запропонованих планувальників. Для цього ми використовуємо такі параметри:

Ми задаємо для [1-2] невеликі значення, щоб у разі виконання запитів в одному потоці очікування не було занадто тривалим.

2.8.1. з планувальником [Schedulers.io]

Можна відзначити наступні моменти:

• відповіді надходять не в тому порядку, в якому надходили запити (див. idClient);
• кожен запит виконувався в окремому потоці;
• графічний інтерфейс цього разу вже не застиг:
  • можна переходити з однієї вкладки на іншу;
  • видно, як надходять дані;
  • не встигаємо побачити кнопку [Annuler], оскільки виконання відбувається надто швидко. Ми виділимо її в іншому тесті;

2.8.2. за допомогою планувальника [Schedulers.computation]

Можна відзначити наступне:

• відповіді надходять не в тому порядку, в якому надходили запити (див. idClient);
• запити виконувалися у 8 потоках;
• потік № 3 використовувався для запитів 8 і 0;
• потік № 4 використовувався для запитів 9 та 1;
• решта запитів виконувалися кожен у своєму потоці;

Планувальник [Schedulers.computation] використовує стільки потоків, скільки є ядер на використовуваній машині. Ця інформація отримується за допомогою виразу [Runtime.getRuntime().availableProcessors()].

2.8.3. з планувальником [Schedulers.newThread]

Функціонування аналогічне до роботи планувальника [Schedulers.io].

2.8.4. з планувальниками [Schedulers.trampoline, Schedulers.immediate]

Функціонування є синхронним. Усі запити виконуються у потоці циклу подій. Не слід узагальнювати цей результат, але можна сказати, що саме в цьому конкретному прикладі обидва планувальники працювали синхронно.

2.9. Граничні випадки

У цьому прикладі ми будемо працювати з планувальниками, які підтримують асинхронну роботу. Спочатку збільшимо кількість запитів до 100 із використанням планувальника [Schedulers.computation], який тут працює з 8 потоками. Отримаємо такий результат:

• у [1] кнопка [Annuler] присутня та доступна для використання (асинхронна робота);

Тепер дочекаємося завершення виконання:

З [2] видно, що виконання 100 запитів зайняло приблизно 4 секунди (на 8 потоках).

Тепер виконаємо ті самі 100 запитів за допомогою планувальника [Schedulers.newThread], який виконує кожен запит в окремому потоці:

У [1] бачимо, що виконання 100 запитів (на 100 потоках) зайняло півсекунди. Отже, це значно швидше, ніж із планувальником [Schedulers.computation].

Тепер виконаємо 800 запитів за тих самих умов, знову використовуючи планувальник [Schedulers.newThread]. Отримаємо такі результати:

800 запитів виконуються приблизно за 1 секунду.

Коли ми збільшуємо цю кількість (понад 2500 запитів на моїй машині — виконаних за 1,5 с — ця кількість, звісно, дуже залежить від робочого середовища на момент виконання), ми в кінцевому підсумку отримуємо таке виключення:

Отже, маємо переповнення стека. Тести показують, що робота планувальника [Schedulers.newThread] не є детермінованою. Може виникнути вищезазначене виключення, можна провести нові випробування, а потім повернутися до конфігурації, яка спричинила виключення, і воно більше не виникне.

2.10. Conclusion

Ми продемонстрували приклад використання бібліотеки Rx. Підсумуємо те, що ми дізналися:

Ми виходили з такої архітектури:

• у [4] шар [swing] виконував синхронні виклики до шару [service];
• у [5], шар [swing] здійснював асинхронні виклики до шару [rxService], який, у свою чергу, синхронно викликав шар [service] через шар [6];

Перше, що ми помітили, — це те, що бібліотека Rx давала змогу легко створити асинхронний інтерфейс [rxService] на основі синхронного інтерфейсу [service] (див. розділ 2.4). Це важливий висновок, оскільки це означає, що синхронний додаток можна легко перетворити на асинхронний.

У шарі [swing] було написано два окремі методи:

• один — для синхронних викликів сервісу (див. розділ 2.4);
• інший — для асинхронних викликів (див. параграф 2.7);

Написання асинхронних викликів виявилося значно складнішим, ніж синхронних. Проте ті, хто займався паралельним програмуванням із синхронізацією декількох потоків, переконаються, що рішення на основі Rx простіше в реалізації та дозволяє уникнути всіх складних проблем синхронізації та взаємодії між потоками. Під час написання коду ми виділили такі важливі моменти:

• тип [Observable] позначає потік подій (значень), які можуть бути (але не обов’язково) асинхронними та які можна спостерігати;
• тип [Subscriber] позначає підписника на тип [Observable];
• тип [Subscription] позначає підписку, тобто зв’язок між [Subscriber] та [Observable];
• тип [Observable] допускає оператори [mergeWith, empty, subscribeOn, observeOn, ...], які здебільшого генерують спостережувані величини. Ці оператори слугують для налаштування спостережуваної величини перед її виконанням:
  • що саме потрібно спостерігати;
  • потік, на якому виконується об’єкт спостереження;
  • потік, у якому підписник отримує дані від об’єкта спостереження;
• розрізняють два типи об’єктів спостереження: [froid / cold] та [chaud / hot]. «Холодний» об’єкт спостереження повністю виконується при підключенні кожного нового підписника. Якщо кожне виконання генерує однакові дані, кожен новий підписник отримує ті самі дані, що й попередній. «Гарячий» обсервабель зазвичай генерує дані безперервно. Коли підписник підписується, він отримує дані, що були відправлені з моменту його підписки. Він не отримує даних, які могли бути відправлені раніше. У нашому прикладі обсервабель є «холодним»: він повністю перевиконується при кожному новому підписнику.

Тепер, коли ми розглянули приклад, який продемонстрував переваги бібліотеки Rx, ми детальніше ознайомимося з нею.

Бібліотека Rx містить численні методи, сигнатури яких містять генеричні параметри. Ми коротко нагадаємо про ці сигнатури (параграф 3). Параметри цих методів здебільшого є функціональними інтерфейсами (Java 8), тобто інтерфейсами, що мають лише один метод. Отже, фактичними параметрами мають бути екземпляри цих інтерфейсів. До Java 8 зазвичай реалізовували інтерфейс за допомогою анонімного класу. У Java 8, якщо інтерфейс є функціональним, лаконічніше реалізувати його за допомогою лямбда-функції. Тому ми розглянемо їх (параграф 4). Після цього ми розглянемо клас [Stream] (параграф 5), який дозволяє обробляти колекції Java за допомогою лямбда-функцій. Цей клас цікавий тим, що клас [Observable] з RxJava запозичує у нього:

• деякі методи;
• той самий спосіб ланцюгового виклику методів для обробки одного й того самого об’єкта спостереження;

Далі ми розглянемо функціональні інтерфейси, специфічні для бібліотеки RxJava (параграф 6). Далі ми перейдемо до основних елементів бібліотеки Rx [Observable, Subscriber, Subscription, opérateurs] (параграф 7). Клас [Observable] має кілька десятків операторів, які самі по собі перевантажуються кілька разів. Спочатку це створює значну складність, оскільки ці оператори та їхні перевантаження іноді відрізняються лише однією деталлю, і без досвіду важко визначити, який саме оператор слід використовувати. Ми розглянемо лише обмежену кількість операторів, а їхні перевантаження здебільшого ігноруватимемо.

Уся попередня частина буде реалізована з використанням бібліотеки RxJava у простих консольних додатках. Після ознайомлення з бібліотекою RxJava ми будемо використовувати її у двох типах графічних додатків:

• у розділі 8 ми повернемося до прикладу програми на Swing, щоб розглянути її детальніше. Тоді ми використовуватимемо бібліотеку RxSwing;
• у розділі 9 ми створимо додаток для Android із використанням бібліотеки RxAndroid;

Коли все це буде зроблено, читач отримає інструменти, щоб діяти самостійно. Ймовірно, знадобиться деякий час, перш ніж він зможе інтуїтивно користуватися бібліотекою Rx. Я вважаю цю бібліотеку особливо цікавою. Однак мені вона здалася складною для розуміння, і процес навчання зайняв чимало часу. Сподіваюся, що цей документ допоможе читачеві скоротити цей час. Мені здається, що це того варте.