7. Biblioteka RxJava
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 uruchamianie 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.
Biblioteka RxJava jest szczególnie dobrze dostosowana do architektury omówionej w akapicie 2 sekcji „Odkrycie”, którą przytaczamy tutaj:
- w [1] warstwa usług świadczy usługi, z których niektóre wymagają dłuższego czasu realizacji (na przykład żądania sieciowe);
- warstwa usług jest wywoływana przez interfejs graficzny [1] (Swing, Android, JavaFx). Jeśli warstwa usług jest wykonywana w tym samym wątku co metoda [swing], która z niej korzysta, interfejs graficzny zawiesza się (nie reaguje) podczas oczekiwania na wynik usługi;
- w [2] cienka warstwa adaptacyjna zaimplementowana przy użyciu RxJava pozwala przedstawić warstwie graficznej asynchroniczną implementację tej samej usługi: może ona być wykonywana w innym wątku niż metoda warstwy graficznej, która ją wywołuje. W tym przypadku interfejs graficzny [3] pozostaje responsywny: użytkownik może nadal z nim współpracować, na przykład wysłać nowe żądanie sieciowe równolegle do pierwszego, a co najważniejsze, można mu zaoferować możliwość anulowania zbyt długotrwałych operacji, co jest niemożliwe, gdy interfejs graficzny jest zablokowany;
- wywołanie [4] jest synchroniczne, podczas gdy wywołanie [5-6] jest asynchroniczne;
W tej architekturze warstwa [2] udostępnia usługi zwracające typy Observable<T>, które mogą być subskrybowane przez metody warstwy graficznej [3]. Usługa warstwy [2] dostarcza następnie swoje wyniki pojedynczo, a warstwa [3] może reagować na każdy z nich, na przykład aktualizując jeden lub więcej elementów interfejsu graficznego.
Klasa Observable<T> posiada kilkadziesiąt metod. Jest to jedna z trudności związanych z tą biblioteką: jest ona bardzo bogata i trudno jest ogarnąć wszystkie jej możliwości. Przedstawimy tutaj niektóre z nich. Opanowanie pozostałych metod nastąpi z czasem.
7.1. Tworzenie obserwowalnych obiektów i subskrybowanie ich
7.1.1. Przykład-01: metoda [Observable.from]
 |
Rozważmy następujący kod:
package dvp.rxjava.observables;
import rx.Observable;
import rx.functions.Action0;
import rx.functions.Action1;
import java.util.Arrays;
public class Exemple01 {
public static void main(String[] args) {
// wielkości obserwowalne typu całkowitego
Observable<Integer> obs1 = Observable.from(Arrays.asList(1, 2, 3));
obs1.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.printf("next : %s%n", integer);
}
}, new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
System.out.println(throwable);
}
}, new Action0() {
@Override
public void call() {
System.out.println("completed");
}
});
}
}
- wiersz 12: tworzymy typ Observable<Integer> na podstawie listy liczb całkowitych.
Klasa Observable<T> jest strumieniem elementów typu T, które można obserwować – najlepiej asynchronicznie, ale niekoniecznie – w miarę ich generowania. Jej definicja jest następująca:
 |
Jak już wspomniano, klasa Observable<T> posiada kilkadziesiąt metod. Niektóre z nich są podobne do metod klasy Stream<T> omówionej w akapicie 5. Dokumentacja klasy RxJava zawiera „diagramy marmurowane” [2], które ilustrują działanie tych metod:
- linia 3 przedstawia emisje obserwowalnej w funkcji czasu;
- metoda [4] jest stosowana do elementów emitowanych przez obserwowalną. Zazwyczaj generuje ona nową obserwowalną;
- wiersz 5 przedstawia uzyskaną nową obserwowalną;
Metoda [Observable.from] ma następującą sygnaturę:
 |
Metoda statyczna [Observable.from] pozwala utworzyć obiekt typu Observable<T> na podstawie zbioru elementów typu T. Jest to bardzo prosty sposób na rozpoczęcie pracy z obserwowalnymi. Wiersz:
Observable<Integer> obs1 = Observable.from(Arrays.asList(1, 2, 3));
wyemituje zatem trzy elementy. Nie wyemituje ich jednak natychmiast. Wyemituje je w całości za każdym razem, gdy zgłosi się obserwator. Nazywa się to obserwowalnością „zimną”. Obserwowalność ponownie emituje swoje elementy dla każdego nowego subskrybenta.
Poprzednią instrukcję można traktować jako operację konfiguracji obserwowalnego. Jest on konfigurowany jednokrotnie i wykonywany n razy, jeśli zgłosi się n subskrybentów.
Jak subskrybować?
Jednym ze sposobów jest użycie metody [Observable.subscribe], której definicja zastosowana w tym przypadku jest następująca:
 |
- pierwszy parametr [Action1<T> onNext] (patrz paragraf 6.2) tej metody to metoda, która ma zostać wykonana, gdy obserwowalny wyemituje nowy element T;
- drugi parametr metody [Action1<Throwable> onError] to metoda, która ma zostać wykonana, gdy obserwowalny generuje wyjątek;
- trzeci parametr metody [Action0 onComplete] (patrz punkt 6.1) to metoda, która ma zostać wykonana, gdy obserwowalny generuje wyjątek;
- metoda zwraca typ [Subscription];
Typ [Subscription] reprezentuje subskrypcję obiektu obserwowalnego. Jego definicja jest następująca:
 |
Znaczenie tego interfejsu [1] polega na jego metodzie [2], która umożliwia anulowanie subskrypcji.
W naszym przykładzie kod subskrypcji obserwowalnej wygląda następująco:
obs1.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.printf("next : %s%n", integer);
}
}, new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
System.out.println(throwable);
}
}, new Action0() {
@Override
public void call() {
System.out.println("completed");
}
});
- wiersz 1: wynik typu [Subscription] jest ignorowany;
- wiersze 1–15: trzy parametry są instancjami klas anonimowych. Wykorzystamy również wyrażenia lambda. Zaletą klas anonimowych jest to, że wyraźnie widać typy danych oczekiwane przez jedyną metodę tych klas;
- wiersze 2–5: implementacja pierwszego parametru typu [Action1<Integer>];
- wiersze 6–10: implementacja drugiego parametru typu [Action1<Throwable>];
- wiersze 11–15: implementacja trzeciego parametru typu [Action0];
Cały kod wygląda następująco:
package dvp.rxjava.observables;
import rx.Observable;
import rx.functions.Action0;
import rx.functions.Action1;
import java.util.Arrays;
public class Exemple01 {
public static void main(String[] args) {
// obserwowalne typu całkowitego
Observable<Integer> obs1 = Observable.from(Arrays.asList(1, 2, 3));
// subskrypcja
obs1.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.printf("next : %s%n", integer);
}
}, new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
System.out.println(throwable);
}
}, new Action0() {
@Override
public void call() {
System.out.println("completed");
}
});
}
}
Obserwowalna w wierszu 12 zaczyna emitować swoje 3 elementy, gdy tylko wywołana zostanie metoda [subscribe] w wierszu 14. Od tego momentu:
- przy każdym wyemitowanym elemencie wykonywane są wiersze 15–18.
- po wysłaniu wszystkich trzech elementów wykonywane są wiersze 24–29;
- wiersze 19–24 nigdy nie zostaną wykonane, ponieważ obserwowalny nie generuje tutaj wyjątku;
Domyślnie obserwowalny i obserwator działają w tym samym wątku. Istnieje kilka predefiniowanych obserwowalnych, które działają w innym wątku niż wątek główny (w tym przypadku wątek metody main), ale w przypadku większości z nich tak nie jest. W tym przypadku wszystko dzieje się więc w wątku metody [main]:
- obserwowalna wysyła element 1;
- wykonywane są linie 15–18 i wyświetlają ten element;
- obserwowalny emituje element 2;
- wykonywane są linie 15–18 i wyświetlany jest ten element;
- obserwowalny wysyła element 3;
- wykonywane są wiersze 15–18 i wyświetlany jest ten element;
- obserwowalny wysyła powiadomienie [completed];
- wykonywane są wiersze 24–29;
Tak wyglądają uzyskane wyniki:
Klasa [Exemple02] jest pochodną klasy [Exemple01], przy czym tym razem jako parametry metody [Observable.subscribe] wykorzystuje funkcje lambda:
package dvp.rxjava.observables;
import java.util.Arrays;
import rx.Observable;
public class Exemple02 {
public static void main(String[] args) {
// wielkości obserwowalne typu całkowitego
Observable<Integer> obs1 = Observable.from(Arrays.asList(1, 2, 3));
// subskrypcja
obs1.subscribe(
(integer) -> System.out.printf("next : %s%n", integer),
(th) -> System.out.println(th),
() -> System.out.println("completed"));
}
}
7.1.2. Przykład 03: klasa Observer
 |
Metoda [Observable.subscribe], umożliwiająca subskrypcję obserwowalnego, ma różne wersje, w tym następującą:
package dvp.rxjava.observables;
import java.util.Arrays;
import rx.Observable;
import rx.Observer;
public class Exemple03 {
public static void main(String[] args) {
// wielkości obserwowalne typu całkowitego
Observable<Integer> obs1 = Observable.from(Arrays.asList(1, 2, 3));
// subskrypcja
obs1.subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
System.out.println("completed");
}
@Override
public void onError(Throwable th) {
System.out.printf("throwable %s", th);
}
@Override
public void onNext(Integer integer) {
System.out.printf("next : %s%n", integer);
}
});
};
}
W wierszu 13, zamiast przekazywać trzy parametry do metody [subscribe], przekazuje się jej następujący typ [Observer]:
 |
Typ [Observer] jest interfejsem zawierającym trzy metody:
- [onNext(T t)], która jest wywoływana za każdym razem, gdy obserwowalny emituje element t;
- [onError(Throwable th)], która jest wywoływana, gdy obserwowalny generuje wyjątek th;
- [onCompleted], która jest wywoływana, gdy obserwowalna wskazuje, że zakończyła emisję;
Działanie kodu jest analogiczne do tego, co wyjaśniono wcześniej. Otrzymujemy następujące wyniki:
7.1.3. Przykład-04: metoda [Observable.create]
 |
Metoda statyczna Observable.create jest zdefiniowana w następujący sposób:
 |
- metoda [create] zwraca typ Observable<T>;
- parametrem metody [create] jest funkcja typu [Observable.OnSubscribe<T>] zdefiniowana w następujący sposób:
 |
Typ [Observable.OnSubscribe<T>] jest interfejsem funkcjonalnym, który sam rozszerza interfejs funkcjonalny [Action1<Subscriber<? super T>>]. Metoda [call] tego interfejsu oczekuje typu [Subscriber] (subskrybent, obserwator), zdefiniowanego w następujący sposób:
 |
W [1] widać, że klasa [Subscriber<T>] implementuje interfejs [Observer<T>] przedstawiony w paragrafie 7.1.2.
Ostatecznie metoda [<T> Observable.create]:
- oczekuje jako parametr instancję typu [Observable.OnSubscribe<T>] posiadającą jedyną metodę o sygnaturze: void call(Subscriber<T> s). Typ [Subscriber<T>] rozszerza typ [Observer<T>] i w związku z tym posiada metody onNext, onError, onCompleted;
- zwraca typ Observable<T>;
Metoda [<T> Observable.create] zwraca skonfigurowaną obserwowalną. Nie nastąpiła jeszcze żadna emisja elementów. Gdy subskrybent [Subscriber<T> s] subskrybuje ten obiekt obserwowalny, wywoływana jest metoda [void call(s)] funkcji przekazanej jako parametr metody [<T> Observable.create]. Jej zadaniem jest wysyłanie elementów t typu T oraz wywoływanie metody [s.onNext(t)] obserwatora przy każdej emisji. Gdy ta metoda zostanie zakończona, należy wywołać metodę [s.onCompleted(t)] obserwatora, a metoda [call] musi zostać zakończona. Jeśli metoda [call] napotka wyjątek th, należy wywołać metodę [s.onError(th)] obserwatora, a metoda [call] powinna zakończyć działanie;
Aby zilustrować to złożone działanie, wykorzystamy następujący kod [Exemple04]:
package dvp.rxjava.observables;
import rx.Observable;
import rx.Subscriber;
import java.util.Random;
public class Exemple04 {
public static void main(String[] args) {
// konfiguracja obserwowalna liczb rzeczywistych
Observable<Double> obs1 = Observable.create(new Observable.OnSubscribe<Double>() {
@Override
public void call(Subscriber<? super Double> subscriber) {
for (int i = 0; i < 3; i++) {
// emisja elementu i
subscriber.onNext(new Random((i + 1)).nextDouble());
}
// koniec transmisji
subscriber.onCompleted();
}
});
// subskrypcja, a zatem emisja
obs1.subscribe((d) -> System.out.printf("onNext %s%n", d), (th) -> System.out.printf("onError %s%n", th),
() -> System.out.println("onCompleted"));
}
}
- wiersz 11: tworzymy obserwowalną, emitującą typy Double;
- wiersze 11–21: parametr metody [create] jest instancjonowany za pomocą anonimowej klasy posiadającej jedyną metodę [call] z wierszy 12–20. Obserwowalny utworzony w wierszu 11 jest gotowy do emisji, ale zacznie emitować dopiero po pojawieniu się obserwatora;
- wiersze 13–21: metoda [call] otrzymuje odwołanie do obserwatora;
- wiersze 14–17: wysłanie 3 elementów do obserwatora;
- wiersz 19: powiadomienie obserwatora o zakończeniu wysyłania;
- wiersze 23–24: subskrypcja obserwowalnej z wiersza 11. Trzy parametry [onNext, onError, onCompleted] metody [subscribe] są implementowane za pomocą trzech funkcji lambda. Ta subskrypcja utworzy subskrybenta [Subscriber<Double>], który zostanie przekazany do metody [call] w wierszu 13. Następnie rozpocznie się wysyłanie elementów;
- wszystko odbywa się w tym samym wątku: obserwowalny i obserwator;
Otrzymujemy następujące wyniki:
Metoda [Observable.create] pozwala utworzyć obserwowalną na podstawie dowolnego zjawiska. To właśnie tę metodę wykorzystaliśmy w akapicie 2 sekcji „Odkrycie”, aby przekształcić interfejs synchroniczny w interfejs asynchroniczny.
7.1.4. Przykład 05: refaktoryzacja [Exemple-04]
 |
Poniższy przykład przedstawia nową wersję metody statycznej [Observable.subscribe]:
package dvp.rxjava.observables;
import rx.Observable;
import rx.Subscriber;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Random;
public class Exemple05 {
public static void main(String[] args) {
// konfiguracja obserwowalnej zmiennej rzeczywistej
Observable<Double> obs1 = Observable.create(new Observable.OnSubscribe<Double>() {
@Override
public void call(Subscriber<? super Double> subscriber) {
showInfos("Observable.call start");
for (int i = 0; i < 3; i++) {
// oczekiwanie
try {
Thread.sleep(500 - i * 100);
} catch (InterruptedException e) {
// błąd
subscriber.onError(e);
}
// działanie
double value = new Random().nextInt(100) * 1.2;
showInfos(String.format("Observable.call onNext(%s)", value));
subscriber.onNext(value);
}
// zakończono
showInfos(String.format("Observable.call onCompleted"));
subscriber.onCompleted();
}
});
// subskrybent
Subscriber<Double> subscriber = new Subscriber<Double>() {
@Override
public void onCompleted() {
showInfos("Subscriber.onCompleted");
}
@Override
public void onError(Throwable e) {
showInfos(String.format("Subscriber.onError (%s)", e));
}
@Override
public void onNext(Double aDouble) {
showInfos(String.format("Subscriber.onNext (%s)", aDouble));
}
};
// subskrypcja
showInfos("avant souscription");
obs1.subscribe(subscriber);
showInfos("après souscription");
}
private static void showInfos(String message) {
System.out.printf("%s ------Thread[%s] ---- Time[%s]%n", message, Thread.currentThread().getName(),
new SimpleDateFormat("ss:SSS").format(new Date()));
}
}
- wiersz 56: nowa wersja metody statycznej [Observable.subscribe] przyjmuje jako parametr typ [Subscriber], który przedstawiliśmy w poprzednim akapicie;
- wiersze 37–52: subskrybent (abonent, obserwator). Implementuje on interfejs Observer wraz z trzema metodami: onNext, onError, onCompleted;
- wiersze 61–64: od tej pory skupimy się na wątkach, w których wykonywane są obserwowalna i jej obserwator;
- wiersz 62: nazwa wątku;
- wiersz 63: aktualny czas wyrażony w sekundach i milisekundach. Pozwoli nam to śledzić w czasie wysyłanie elementów przez obserwowalną oraz ich przetwarzanie przez obserwator;
- kod ten ma taką samą funkcjonalność jak poprzedni. Po prostu go zrefaktoryzowaliśmy;
Uzyskane wyniki są następujące:
- wiersz 1 wyników: przed wierszem 56 kodu nic się jeszcze nie wydarzyło. Obserwowalnik został po prostu skonfigurowany;
- wiersz 2 wyników: wiersz 56 kodu powoduje wywołanie metody [call] z wiersza 15. W wierszu 3 wartość rzeczywista 80,39 jest wysyłana do obserwatora;
- wiersz 4: obserwator odbiera wysłaną liczbę;
- wiersze 5–8: powyższy proces powtarza się dwukrotnie;
- wiersz 9: obserwowalny wysyła powiadomienie o zakończeniu transmisji;
- wiersz 10: obserwator ją odbiera;
- wiersz 11: wyświetlany przez wiersz 57 kodu;
Widać zatem, że tylko wiersz 56 dotyczący subskrypcji spowodował wyświetlenie wierszy 2–10 wyników. Rozpoczynając pracę z biblioteką RxJava, zastanawiamy się, w jaki sposób poszczególne elementy są ze sobą powiązane, a zwłaszcza jakie powiązania istnieją między obserwatorem a obserwowalnym. Widać tutaj, że wiersz 57, czyli subskrypcja obserwowalnego,
- spowodowała wyemitowanie wszystkich elementów obserwowalnego;
- że obserwowalna i obserwator są wykonywane w tym samym wątku;
- że z tego powodu obserwujemy następującą sekwencję: wysłanie elementu i, obserwacja elementu i, wysłanie elementu (i+1), obserwacja elementu (i+1), ...
Przypomnijmy, że emiter czekał przed wysłaniem swoich elementów:
// oczekiwanie
try {
Thread.sleep(500 - i * 100);
} catch (InterruptedException e) {
// błąd
subscriber.onError(e);
}
gdzie i w wierszu 3 oznacza numer transmisji (0<=i<3). Jeśli przyjrzymy się czasom transmisji elementów obserwowalnego:
- wiersze 2, 3: element 0 został wysłany około 500 ms po rozpoczęciu subskrypcji;
- wiersze 3, 5: element 1 został wysłany około 400 ms po elemencie 0;
- wiersze 5, 7: element 2 został wysłany około 300 ms po elemencie 1;
7.2. Wątek wykonawczy, wątek obserwacyjny
7.2.1. Przykład-06: obserwowalna i obserwator w wątku innym niż [main]
 |
Przeprojektowujemy poprzedni przykład w następujący sposób: [Exemple06]:
package dvp.rxjava.observables;
import rx.Observable;
import rx.Subscriber;
import rx.schedulers.Schedulers;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
public class Exemple06 {
public static void main(String[] args) {
// bariera ochronna
CountDownLatch latch = new CountDownLatch(1);
// konfiguracja obserwowalnych wartości rzeczywistych
Observable<Double> obs1 = Observable.create(new Observable.OnSubscribe<Double>() {
@Override
public void call(Subscriber<? super Double> subscriber) {
showInfos("Observable.call start");
for (int i = 0; i < 3; i++) {
// oczekiwanie
try {
Thread.sleep(500 - i * 100);
} catch (InterruptedException e) {
// błąd
subscriber.onError(e);
}
// działanie
double value = new Random().nextInt(100) * 1.2;
showInfos(String.format("Observable.call onNext(%s)", value));
subscriber.onNext(value);
}
// zakończone
showInfos(String.format("Observable.call onCompleted"));
subscriber.onCompleted();
}
});
// subskrybent
Subscriber<Double> subscriber = new Subscriber<Double>() {
@Override
public void onCompleted() {
showInfos("Subscriber.onCompleted");
// opuszczamy barierę
latch.countDown();
}
@Override
public void onError(Throwable e) {
showInfos(String.format("Subscriber.onError (%s)", e));
}
@Override
public void onNext(Double aDouble) {
showInfos(String.format("Subscriber.onNext (%s)", aDouble));
}
};
// kontynuacja konfiguracji obserwowalnej
obs1 = obs1.subscribeOn(Schedulers.computation());
// subskrypcja
showInfos("avant souscription");
obs1.subscribe(subscriber);
// oczekiwanie przed szlabanem
try {
showInfos("début attente barrière");
latch.await();
showInfos("fin attente barrière");
} catch (InterruptedException e1) {
System.out.println(e1);
}
showInfos("après souscription");
}
private static void showInfos(String message) {
System.out.printf("%s ------Thread[%s] ---- Time[%s]%n", message, Thread.currentThread().getName(),
new SimpleDateFormat("ss:SSS").format(new Date()));
}
}
- wiersz 16: tworzymy barierę (semafor) za pomocą obiektu typu [CountDownLatch]. Obiekt ten służy do synchronizacji wątków między sobą. Jest on tutaj zainicjowany wartością 1, którą nazwiemy wartością bariery (lub semafora). Wątek oczekuje na barierę za pomocą operacji:
latch.await();
Wątek zostaje zablokowany, jeśli wartość bariery jest większa od 0. Wątek może zwiększać lub zmniejszać wewnętrzną wartość bariery. W wierszu 48 wartość bariery jest zmniejszana o 1.
- Wiersz 63: obserwowalna jest skonfigurowana tak, aby była wykonywana w wątku dostarczonym przez harmonogram [Schedulers.computation()]. Harmonogram ten może dostarczyć tyle wątków, ile jest rdzeni w maszynie wykonawczej. W rozdziale poświęconym przykładowej aplikacji pokazano wykorzystanie innych harmonogramów (patrz rozdział 2.8);
Zasada działania kodu jest następująca:
- metoda [main] jest wykonywana w wątku głównym (main);
- wiersz 66: uruchamia wysyłanie elementów obserwowalnego. Będą one wysyłane w wątku innym niż wątek główny;
- wiersz 70: wątek główny zostaje zablokowany, ponieważ wartość bariery wynosi 1 (patrz wiersz 16). Będzie mógł kontynuować działanie dopiero wtedy, gdy wartość ta zmieni się na 0. Dzieje się to w wierszu 48. To obserwator obniża barierę, gdy otrzymuje powiadomienie, że obserwowalna zakończyła wysyłanie danych;
Wykonanie daje następujące wyniki:
- wiersz 1: nastąpi subskrypcja;
- wiersz 2: subskrypcja uruchamia metodę [call] w wątku [RxComputationThreadPool-1]. Mamy teraz równoległe wykonywanie z dwoma wątkami;
- wiersz 3: z nieznanych przyczyn wątek [RxComputationThreadPool-1] przekazał kontrolę. Wątek [main] przejmuje wówczas kontrolę i zostaje zablokowany przez barierę (wiersz 70 kodu). Od tego momentu działa już tylko wątek [RxComputationThreadPool-1];
- wiersze 4–11: obserwujemy zachowanie znane z poprzedniego przykładu między obserwowalną wielkością a jej obserwatorem, ale teraz wszystko dzieje się w wątku [RxComputationThreadPool-1];
- wiersze 12–13: obserwator opuścił barierę (wiersz 48 kodu), a wątek [RxComputationThreadPool-1] zakończył działanie. Wątek [main] przejmuje kontrolę i wyświetla dwa komunikaty;
7.2.2. Przykład-07: obserwowalna i obserwator w dwóch różnych wątkach
|
Modyfikujemy poprzedni przykład w następujący sposób:
package dvp.rxjava.observables;
import rx.Observable;
import rx.Subscriber;
import rx.schedulers.Schedulers;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
public class Exemple07 {
public static void main(String[] args) {
// strażnik szlabanu
CountDownLatch latch = new CountDownLatch(1);
// konfiguracja zmiennej obserwowalnej typu liczb rzeczywistych
Observable<Double> obs1 = Observable.create(new Observable.OnSubscribe<Double>() {
@Override
public void call(Subscriber<? super Double> subscriber) {
showInfos("Observable.call start");
for (int i = 0; i < 3; i++) {
// oczekiwanie
try {
Thread.sleep(500 - i * 100);
} catch (InterruptedException e) {
// błąd
subscriber.onError(e);
}
// działanie
double value = new Random().nextInt(100) * 1.2;
showInfos(String.format("Observable.call onNext(%s)", value));
subscriber.onNext(value);
}
// zakończone
showInfos(String.format("Observable.call onCompleted"));
subscriber.onCompleted();
}
});
// subskrybent
Subscriber<Double> subscriber = new Subscriber<Double>() {
@Override
public void onCompleted() {
showInfos("Subscriber.onCompleted");
// opuszczamy barierę
latch.countDown();
}
@Override
public void onError(Throwable e) {
showInfos(String.format("Subscriber.onError (%s)", e));
}
@Override
public void onNext(Double aDouble) {
showInfos(String.format("Subscriber.onNext (%s)", aDouble));
}
};
// kontynuacja konfiguracji obserwowalnej
obs1 = obs1.subscribeOn(Schedulers.computation()).observeOn(Schedulers.computation());
// subskrypcja
showInfos("avant souscription");
obs1.subscribe(subscriber);
// oczekiwanie na podniesienie bariery
try {
showInfos("début attente barrière");
latch.await();
showInfos("fin attente barrière");
} catch (InterruptedException e1) {
System.out.println(e1);
}
showInfos("après souscription");
}
private static void showInfos(String message) {
System.out.printf("%s ------Thread[%s] ---- Time[%s]%n", message, Thread.currentThread().getName(),
new SimpleDateFormat("ss:SSS").format(new Date()));
}
}
Kod jest identyczny jak w poprzednim przykładzie, z wyjątkiem linii 63:
obs1 = obs1.subscribeOn(Schedulers.computation()).observeOn(Schedulers.computation());
która konfiguruje obserwowalną (subscribeOn) i obserwatora (observeOn) tak, aby były wykonywane w jednym z wątków dostarczanych przez harmonogram [Schedulers.computation()].
Otrzymane wyniki są następujące:
Można zauważyć następujące kwestie:
- obserwowalny jest wykonywany w wątku [RxComputationThreadPool-4] (wiersze 3–4, 6, 8–9);
- obserwator działa w wątku [RxComputationThreadPool-3] (wiersze 5, 7, 10–11);
- że działają one niezależnie. Tak więc w wierszach 8–9 obserwowalny wysyła 2 powiadomienia (onNext, onCompleted), zanim obserwator odbierze powiadomienie [onNext] (wiersz 10);
Biblioteka RxJava zajmuje się przekazywaniem danych (emisjami) z wątku obserwowalnego do wątku obserwatora. Programista nie musi się tym martwić.
Widzieliśmy już, jak tworzyć obserwowalne obiekty (Observable.from, Observable.create). Teraz przyjrzyjmy się predefiniowanym obserwowalnym obiektom z biblioteki RxJava.
7.3. Wstępnie zdefiniowane obiekty obserwowalne
7.3.1. Przykład-08: metoda [Observable.range]
 |  |
Od tej pory będziemy używać dedykowanych klas dla obserwowanych procesów i ich obserwatorów. Chodzi o to, aby móc rejestrować ich nazwy, wątki wykonania oraz czasy wykonania, co pozwoli śledzić je w czasie.
Klasa [Process] będzie po prostu obiektem typu Observable, któremu można nadać nazwę. Będzie ona implementować następujący interfejs [IProcess]:
package dvp.rxjava.observables.utils;
import rx.Observable;
public interface IProcess<T> {
// nazwa obserwowalnej wielkości
public String getName();
// wielkość obserwowalna
public Observable<T> getObservable();
}
Interfejs ten może być zaimplementowany przez następującą klasę [Process<T>]:
package dvp.rxjava.observables.utils;
import rx.Observable;
import rx.Scheduler;
public class Process<T> implements IProcess<T>{
// nazwa wielkości obserwowalnej
protected String name;
// proces obserwowany
protected Observable<T> observable;
// konstruktorzy
public Process(String name, Observable<T> observable) {
// lokalne inicjalizacje
this.name = name;
this.observable = observable;
}
// metody pobierające i ustawiające
public String getName() {
return name;
}
public Observable<T> getObservable() {
return observable;
}
}
- wiersz 9: nazwa procesu;
- wiersz 11: obserwowana wielkość;
- wiersze 14–18: konstruktor;
Obserwator zostanie opisany przez następującą klasę [Observateur]:
package dvp.rxjava.observables.utils;
import java.util.concurrent.CountDownLatch;
import java.util.function.Consumer;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import rx.Subscriber;
public class Observateur<T> extends Subscriber<T> {
...
}
- w wierszu 11 klasa Observateur<T> rozszerza klasę Subscriber<T>, którą pokrótce przedstawiliśmy w paragrafie 7.1.3. Wykorzystamy ją jako argument metody [Observable.subscribe]:
// obserwowalne wykonanie (obserwacja)
obs1.subscribe(observateur);
Metoda [Observable.subscribe] użyta w wierszu 2 powyżej ma następującą definicję:
 |
Rola [Subscriber] polega głównie na zarządzaniu elementami wysyłanymi przez obserwowalny obiekt, na który subskrybuje, za pomocą metod interfejsu [Observer]: onNext, onError, onCompleted. Klasa [Subscriber] posiada następujące metody:
 |
W kodzie klasy [Observateur] wykorzystamy metodę [1] oraz isUnsubscribed, aby sprawdzić, czy subskrypcja subskrybenta została anulowana, czy nie. Pełny kod klasy [Observateur<T>] wygląda następująco:
package dvp.rxjava.observables.utils;
import java.util.concurrent.CountDownLatch;
import java.util.function.Consumer;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import rx.Subscriber;
public class Observateur<T> extends Subscriber<T> {
// semafor
private CountDownLatch latch;
// metoda wyświetlania
private Consumer<String> showInfos;
// nazwa obserwatora
private String observerName;
// nazwa obserwowanego procesu
private String processName;
// konstruktory
public Observateur() {
}
public Observateur(String name, CountDownLatch latch, Consumer<String> showInfos, String observedName) {
this.observerName = name;
this.latch = latch;
this.showInfos = showInfos;
this.processName = observedName;
}
// --------------------------- implementacja interfejsu Observer<T>
@Override
public void onCompleted() {
// koniec transmisji
if (!isUnsubscribed()) {
showInfos.accept(String.format("Subscriber [%s,%s].onCompleted", observerName, processName));
}
// koniec blokady wątku głównego
latch.countDown();
}
@Override
public void onError(Throwable e) {
// błąd emisji
if (!isUnsubscribed()) {
showInfos.accept(String.format("Subscriber [%s, %s].onError (%s)", observerName, processName, e));
}
}
@Override
public void onNext(T value) {
// dodatkowa emisja
if (!isUnsubscribed()) {
try {
showInfos.accept(String.format("Subscriber [%s,%s] : onNext (%s)", observerName, processName,
new ObjectMapper().writeValueAsString(value)));
} catch (JsonProcessingException e) {
showInfos.accept(String.format("Subscriber [%s,%s].onNext (%s)", observerName, processName, e));
}
}
}
}
- Oprócz cech charakterystycznych dla klasy Subscriber, obserwator Observateur będzie zawierał następujące informacje:
- wiersz 14: blokada lub semafor, który posłuży do zablokowania głównego wątku do momentu, aż obserwator otrzyma wszystkie elementy wysłane przez obserwowalny obiekt. Odbędzie się to w wierszu 36 kodu, gdy obserwator otrzyma od obserwowalnego obiektu powiadomienie o zakończeniu wysyłania;
- wiersz 16: instancja Consumer<String>, która posłuży do wyświetlenia komunikatu na konsoli;
- wiersz 18: nazwa obserwatora, pozwalająca odróżnić je od siebie, gdy jest ich więcej niż jeden;
- wiersz 20: nazwa obserwowanego procesu;
- wiersze 36, 46, 54: metody [onCompleted, onError, onNext] interfejsu [Observer<T>] zaimplementowanego przez klasę abstrakcyjną [Subscriber<T>]. Klasa ta nie implementuje tych metod. Należy to zatem zrobić w jej klasach potomnych. Zanim wykonamy jakiekolwiek działania w tych metodach, sprawdzamy, czy obserwator nie został wyrejestrowany z obserwowalnego obiektu, który obserwuje;
- wiersz 59: metoda [onNext] obserwatora zapisuje ciąg znaków jSON z otrzymanego elementu. Pozwoli nam to wyświetlać różne typy elementów;
Mając to na uwadze, przyjrzyjmy się nowej metodzie klasy Observable, a mianowicie metodzie [range]:
 |
Obserwowalna Observable.range(n,m) generuje (m) liczb całkowitych z przedziału od n do n+m-1. Analizujemy ją za pomocą następującego kodu [Exemple08]:
package dvp.rxjava.observables.exemples;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.function.Consumer;
import dvp.rxjava.observables.utils.Observateur;
import rx.Observable;
import rx.schedulers.Schedulers;
public class Exemple08 {
public static void main(String[] args) throws InterruptedException {
// liczba obserwatorów
final int nbObservateurs = 2;
// semafor
CountDownLatch latch = new CountDownLatch(nbObservateurs);
// konfiguracja obserwowalna
Observable<Integer> obs1 = Observable.range(15, 3).subscribeOn(Schedulers.computation());
// obserwowalne wykonanie (obserwacja)
showInfos.accept("main : début observation");
for (int i = 0; i < nbObservateurs; i++) {
obs1.subscribe(new Observateur<>(String.format("observateur[%d]", i), latch, showInfos,"obs1"));
}
// oczekiwanie
showInfos.accept("main : attente fin observation");
latch.await();
// koniec
showInfos.accept("main : fin observation");
}
// wyświetlenia
static Consumer<String> showInfos = message -> System.out.printf("%s ------Thread[%s] ---- Time[%s]%n", message,
Thread.currentThread().getName(), new SimpleDateFormat("ss:SSS").format(new Date()));
}
- wiersz 16: wykorzystamy dwóch obserwatorów;
- wiersz 19: semafor jest inicjowany na wartość dwa, ponieważ każdy obserwator zostanie umieszczony w innym wątku. Wątek główny będzie więc musiał poczekać na zakończenie działania obu wątków obserwacyjnych;
- wiersz 22: konfigurujemy obserwowalną tak, aby była wykonywana w wątku harmonogramu [Schedulers.computation()]. Obserwator będzie znajdował się w tym samym wątku co obserwowalna;
- wiersze 25–27: subskrybujemy dwa obserwatory do obserwowalnej wartości. Spowoduje to uruchomienie pełnego przebiegu tej wartości dla każdego z obserwatorów: zostaną wysłane liczby całkowite 15, 16 i 17;
- wiersz 30: wątek główny czeka na zakończenie działania obserwatorów;
Uzyskane wyniki są następujące:
- wiersz 2: wątek główny jest zablokowany w oczekiwaniu na zakończenie działania obu obserwatorów;
- wiersze 3–4: widać, że obserwator 0 znajduje się w wątku [RxComputationThreadPool-1], a obserwator 1 w wątku [RxComputationThreadPool-2];
- wiersze 3–10: widać, że obaj obserwatorzy otrzymują dokładnie te same elementy;
Wykorzystamy tak zdefiniowaną klasę Observateur do zilustrowania zachowania innych typów obserwowalnych.
7.3.2. Przykład-09: metody Observable.[interval, take, doNext]
 |
 |
Ten przykład ilustruje wykorzystanie obserwowalnej Observable.interval (długi interwał, jednostka TimeUnit), która generuje liczby całkowite typu long w regularnych odstępach czasu. Należy zwrócić uwagę na punkt [1]: domyślnie obserwowalna [Observable.interval] działa w jednym z wątków harmonogramu [Schedulers.computation].
Kod będzie wyglądał następująco:
package dvp.rxjava.observables.exemples;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import dvp.rxjava.observables.utils.Observateur;
import rx.Observable;
public class Exemple09 {
public static void main(String[] args) throws InterruptedException {
// liczba obserwatorów
final int nbObservateurs = 2;
// semafor
CountDownLatch latch = new CountDownLatch(nbObservateurs);
// konfiguracja obserwowalna
Observable<Long> obs1 = Observable.interval(500L, TimeUnit.MILLISECONDS).take(3)
.doOnNext(l -> showInfos.accept(l.toString()));
// obserwowalne wykonanie (obserwacja)
showInfos.accept("main : début observation");
for (int i = 0; i < nbObservateurs; i++) {
obs1.subscribe(new Observateur<>(String.format("observateur [%d]", i), latch, showInfos,
"obs1"));
}
// oczekiwanie
showInfos.accept("main : attente fin observation");
latch.await();
// koniec
showInfos.accept("main : fin observation");
}
// wyświetlenia
static Consumer<String> showInfos = message -> System.out.printf("%s ------Thread[%s] ---- Time[%s]%n", message,
Thread.currentThread().getName(), new SimpleDateFormat("ss:SSS").format(new Date()));
}
- wiersz 22: obserwowalna generuje liczby całkowite typu long co 500 milisekund. Ciąg zaczyna się od liczby 0;
- wiersz 22: ta obserwowalna generuje nieskończoną liczbę wartości. Metoda [Observable.take(n)] tworzy nową obserwowalną, która zachowuje tylko pierwsze n wygenerowanych elementów;
 |
Wróćmy do kodu obserwowalnego:
Observable<Long> obs1 = Observable.interval(500L, TimeUnit.MILLISECONDS).take(3)
.doOnNext(l -> showInfos.accept(l.toString()));
W wierszu 2 metoda [Observable.doOnNext] jest wykonywana za każdym razem, gdy obserwowalna generuje nowy element. Często wykorzystuje się to do rejestrowania informacji w dzienniku. W tym przypadku chcemy rejestrować datę generowania elementów, aby sprawdzić, czy interwał 500 milisekund jest prawidłowo przestrzegany. Metoda [Observable.doOnNext] nie modyfikuje obserwowalnego, do którego jest zastosowana. Jej definicja jest następująca:
 |
Wykonanie daje następujące wyniki:
- wiersze 3, 7 i 11: widać, że interwał emisji jest w przybliżeniu bliski 500 ms;
- Obydwaj obserwatorzy działają oczywiście w dwóch różnych wątkach, mimo że obserwowalny nie został skonfigurowany do działania z konkretnym harmonogramem. Jest to domyślne działanie obserwowalnego [Observable.interval], które tutaj obserwujemy;
7.3.3. Przykłady-10/12: metody Observable.[error, empty, never]
 |  |
Od tej pory będziemy bardziej zwięźli w ilustrowaniu metod klasy [Observable]. Poprzedni kod wyglądał następująco:
package dvp.rxjava.observables;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import rx.Observable;
public class Exemple09 {
public static void main(String[] args) throws InterruptedException {
// liczba obserwatorów
final int nbObservateurs = 2;
// semafor
CountDownLatch latch = new CountDownLatch(nbObservateurs);
// konfiguracja obserwowalna
Observable<Long> obs1 = Observable.interval(500L, TimeUnit.MILLISECONDS).take(3)
.doOnNext(l -> showInfos.accept(l.toString()));
// obserwowalne wykonanie (obserwacja)
showInfos.accept("main : début observation");
for (int i = 0; i < nbObservateurs; i++) {
obs1.subscribe(new Observateur<>(String.format("observateur [%d]", i), latch, showInfos,
"obs1"));
}
// oczekiwanie
showInfos.accept("main : attente fin observation");
latch.await();
// koniec
showInfos.accept("main : fin observation");
}
// wyświetlacze
static Consumer<String> showInfos = message -> System.out.printf("%s ------Thread[%s] ---- Time[%s]%n", message,
Thread.currentThread().getName(), new SimpleDateFormat("ss:SSS").format(new Date()));
}
Kod ten został już wykorzystany w poprzednim przykładzie. Zmieniły się jedynie wiersze 21–22. Większą część tego kodu wyodrębnimy zatem do następującej klasy [ProcessUtils]:
package dvp.rxjava.observables.utils;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.function.Consumer;
import rx.Observable;
public class ProcessUtils {
@SafeVarargs
public static void subscribe(int nbObservateurs, IProcess<?>... processes) throws InterruptedException {
// semafor
CountDownLatch latch = new CountDownLatch(nbObservateurs * processes.length);
// wykonanie obserwowalne (obserwacja)
showInfos.accept("main : début observation");
for (int i = 0; i < nbObservateurs; i++) {
for (IProcess<?> process : processes) {
Observable<?> obs = process.getObservable();
obs.subscribe(new Observateur<>(String.format("observateur[%d]", i), latch, showInfos, process.getName()));
}
}
// oczekiwanie
showInfos.accept("main : attente fin observation");
latch.await();
// koniec
showInfos.accept("main : fin observation");
}
// wyświetlenia
static Consumer<String> showInfos = message -> System.out.printf("%s ------Thread[%s] ---- Time[%s]%n", message,
Thread.currentThread().getName(), new SimpleDateFormat("ss:SSS").format(new Date()));
}
- wiersz 13: metoda przyjmuje dwa parametry:
- nbObservateurs: liczba obserwatorów procesów przekazanych jako drugi parametr;
- processes: procesy (nazwane obserwowalne) do obserwacji. Dzięki notacji [IProcess<?>] procesy będą mogły generować elementy różnych typów;
- wiersz 16: semafor musi przejść na zielone światło, gdy wszyscy obserwatorzy zakończą wszystkie swoje obserwacje. Wartość początkowa semafora wynosi zatem iloczyn liczby obserwatorów i liczby obserwacji;
- wiersze 20–25: każdy obserwator jest subskrybowany na wszystkie procesy, które należy obserwować;
- wiersz 23: pobieramy obserwowalną wielkość z procesu (patrz paragraf 7.3.1);
- wiersz 23: przypisuje się do niej obserwatora. Przekazuje się mu 4 informacje:
- jego nazwę;
- semafor, który musi zmniejszyć, gdy otrzyma powiadomienie o zakończeniu emisji obserwowalnej, którą obserwuje;
- metodę, której ma użyć, gdy chce zapisać informacje w konsoli;
- nazwę procesu, który będzie obserwował;
Po zdefiniowaniu tych klas przykład 10 będzie wyglądał następująco:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.schedulers.Schedulers;
public class Exemple10 {
public static void main(String[] args) throws InterruptedException {
// konfiguracja obserwowalna
Observable<?> obs = Observable.error(new RuntimeException("Erreur !!!")).subscribeOn(Schedulers.computation());
// wykonanie (obserwacja) obserwowalne
ProcessUtils.subscribe(2,new Process<>("process1", obs));
}
}
W wierszu 11 statyczna metoda [Observable.error] jest zdefiniowana w następujący sposób:
 |
Wiersz 8 konfiguruje zatem obserwowalny obiekt, który po prostu generuje wyjątek skierowany do metody [onError] swoich subskrybentów. Wykonanie daje następujące wyniki:
main : début observation ------Thread[main] ---- Time[22:618]
main : attente fin observation ------Thread[main] ---- Time[22:636]
Subscriber[observateur[1], process1].onError (java.lang.RuntimeException: Erreur !!!) ------Thread[RxComputationThreadPool-2] ---- Time[22:638]
Subscriber[observateur[0], process1].onError (java.lang.RuntimeException: Erreur !!!) ------Thread[RxComputationThreadPool-1] ---- Time[22:638]
W wierszach 3 i 4 metoda [onError] obu subskrybentów otrzymała wyjątek wygenerowany przez obserwowalną.
To wykonanie ma jedną szczególną cechę: metody [onCompleted] obu obserwatorów nie zostały wywołane. W rezultacie bariera nie została opuszczona, a wątek główny pozostaje zablokowany w statycznej metodzie [ProcessUtils.subscribe] w następującej linii 3:
// oczekiwanie
showInfos.accept("main : attente fin observation");
latch.await();
// koniec
showInfos.accept("main : fin observation");
Widać tutaj, że w przypadku błędu obserwowalnej metoda [onCompleted] subskrybentów nie jest wywoływana. Modyfikujemy zatem metodę [Observateur.onError] w następujący sposób:
@Override
public void onError(Throwable e) {
// błąd transmisji
if (!isUnsubscribed()) {
showInfos.accept(String.format("Subscriber[%s, %s].onError (%s)", observerName, processName, e));
}
// koniec blokady wątku głównego
latch.countDown();
}
Dodajemy linie 7–8, aby usunąć blokadę w przypadku błędu obserwowalnej. Po wprowadzeniu tego nowego kodu wykonanie daje następujące wyniki:
main : début observation ------Thread[main] ---- Time[40:750]
main : attente fin observation ------Thread[main] ---- Time[40:764]
Subscriber[observateur[0], process1].onError (java.lang.RuntimeException: Erreur !!!) ------Thread[RxComputationThreadPool-1] ---- Time[40:766]
Subscriber[observateur[1], process1].onError (java.lang.RuntimeException: Erreur !!!) ------Thread[RxComputationThreadPool-2] ---- Time[40:766]
main : fin observation ------Thread[main] ---- Time[40:767]
Otrzymujemy wiersz 5, którego wcześniej nie mieliśmy.
Przykład 11 będzie wyglądał następująco:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
public class Exemple11 {
public static void main(String[] args) throws InterruptedException {
// konfiguracja obserwowalna
Observable<?> obs1 = Observable.empty();
// wykonanie (obserwacja) obserwowalne
ProcessUtils.subscribe(2,new Process<>("process1",obs1));
}
}
W wierszu 10 metoda statyczna [Observable.empty] tworzy obserwowalną, która nie emituje żadnych elementów. Emituje jedynie powiadomienie o zakończeniu emisji;
 |
Wykonanie kodu z powyższego przykładu daje następujące wyniki:
- wiersze 2 i 3: widać, że obaj obserwatorzy otrzymują powiadomienie o zakończeniu emisji, nie otrzymując wcześniej żadnych elementów.
Można się zastanawiać, do czego ta metoda może służyć. Można jej używać analogicznie do kolekcji, która początkowo jest pusta, a do której następnie dodawane są elementy:
W wierszu 3 łączymy początkową obserwowalną obs (wiersz 1) z innymi obserwowalnymi.
Przykład 12 ilustruje metodę statyczną [Observable.never]:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
public class Exemple12 {
public static void main(String[] args) throws InterruptedException {
// konfiguracja obserwowalna
Observable<?> obs1 = Observable.never();
// wykonanie (obserwacja) obserwowalne
ProcessUtils.subscribe(2,new Process<>("process1",obs1));
}
}
Metoda statyczna [Observable.never] tworzy obserwowalną, która nigdy nie emituje:
 |
Wykonanie tego przykładu daje następujące wyniki:
W wierszu 2 główny wątek czeka w nieskończoność. Żadna obserwowalna nie wysyła bowiem powiadomienia [onCompleted], które pozwala przełączyć semafor (barierę) na zielone światło (opuszczenie bariery).
7.4. Multi-threading
7.4.1. Przykład 13: wątek działania, wątek obserwacji
W punkcie 7.1.3 utworzyliśmy obserwowalny obiekt za pomocą metody statycznej [Observable.create]:
|
- metoda [create] zwraca typ Observable<T>;
- parametrem metody [create] jest funkcja typu [Observable.OnSubscribe<T>] zdefiniowana w następujący sposób:
|
Typ [Observable.OnSubscribe<T>] jest interfejsem funkcjonalnym, który sam rozszerza interfejs funkcjonalny [Action1<Subscriber<? super T>>]. Metoda [call] tego interfejsu oczekuje typu [Subscriber] (subskrybent, obserwator). W dalszej części niniejszego dokumentu będziemy czasami nazywać typ [Observable.OnSubscribe<T>] akcją. Utworzymy niestandardowe akcje, które będą miały nazwy. Będą to instancje następującego interfejsu [IProcessAction]:
 |
package dvp.rxjava.observables.utils;
import rx.Observable;
public interface IProcessAction<T> extends Observable.OnSubscribe<T> {
// działanie ma nazwę
public String getName();
}
- wiersz 5: interfejs [IProcessAction<T>] posiada wszystkie cechy interfejsu [Observable.OnSubscribe<T>];
- wiersz 8: posiada ona dodatkowo metodę [getName], która zwraca nazwę instancji implementującej interfejs;
Wykorzystamy następującą akcję o nazwie [ProcessAction01]:
package dvp.rxjava.observables.utils;
import java.util.Random;
import rx.Subscriber;
import rx.functions.Func1;
public class ProcessAction01<T> implements IProcessAction<T> {
// dane
private String name;
private int nbValues;
private Func1<Integer, T> func1;
// konstruktorzy
public ProcessAction01(String name, int nbValues, Func1<Integer, T> func1) {
this.name = name;
this.nbValues = nbValues;
this.func1 = func1;
}
@Override
public void call(Subscriber<? super T> subscriber) {
ProcessUtils.showInfos.accept(String.format("Observable (%s) call start", getName()));
for (int i = 0; i < nbValues; i++) {
// oczekiwanie
try {
Thread.sleep(new Random().nextInt(500));
} catch (InterruptedException e) {
// błąd
ProcessUtils.showInfos.accept(String.format("Observable (%s) onError", getName()));
subscriber.onError(e);
}
// wysłanie elementu
T value = func1.call(i);
ProcessUtils.showInfos.accept(String.format("Observable (%s,%s) onNext (%s)", getName(), i, value));
subscriber.onNext(value);
}
// zakończono
ProcessUtils.showInfos.accept(String.format("Observable (%s) onCompleted", getName()));
subscriber.onCompleted();
}
@Override
public String getName() {
return name;
}
}
- wiersz 8: klasa [ProcessAction01<T>] implementuje interfejs [IProcessAction<T>], a tym samym interfejs [Observable.OnSubscribe<T>];
- wiersz 11: nazwa akcji;
- wiersz 12: liczba wartości do wysłania;
- wiersz 13: instancja typu [Func1<Integer, T>], która na podstawie liczby całkowitej tworzy typ T, który zostanie wyemitowany przez obserwowalną (wiersze 35 i 37);
- wiersze 16–20: do konstruktora przekazywane są nazwa akcji, liczba wartości do wysłania oraz funkcja wysyłania;
- wiersze 23–42: kod procesu;
- wiersz 23: metoda [call] przyjmuje jako parametr subskrybenta obserwowalnego powiązanego z procesem;
- wiersz 28: proces emituje swoje elementy po oczekiwaniu o losowej długości;
- wiersz 32: wysłanie błędu;
- wiersz 37: normalne wysłanie;
- wiersz 41: wysłanie powiadomienia o zakończeniu wysyłania;
- wiersze 25–38: akcja wysyła rzeczywiste wartości nbValues po losowym czasie oczekiwania (wiersz 30);
- wiersz 35: wartość do wysłania jest dostarczana przez funkcję [func1] przekazaną jako parametr do konstruktora (wiersz 16);
Przeprojektowujemy klasę [Process] (patrz paragraf 7.3.1), aby można ją było utworzyć również za pomocą akcji o nazwie. Dodajemy do niej następujący konstruktor:
public Process(IProcessAction<T> na, Scheduler schedulerObserved, Scheduler schedulerObserver) {
// nazwa procesu=nazwa akcji
name = na.getName();
// akcja --> obserwowalna
observable = Observable.create(na);
// wątek wykonania obserwowanego procesu
if (schedulerObserved != null) {
observable = observable.subscribeOn(schedulerObserved);
}
// wątek obserwacyjny obserwatora
if (schedulerObserver != null) {
observable = observable.observeOn(schedulerObserver);
}
}
- w wierszu 1 konstruktor przyjmuje 3 parametry:
- nazwaną akcję, która posłuży do utworzenia obserwowalnego (wiersz 5);
- harmonogram procesu obserwowanego (może to być null);
- harmonogram obserwatora (może to być np. null);
- wiersz 5: obserwowalna jest tworzona na podstawie akcji przekazanej jako parametr;
Poniższy kod [Exemple13] obserwuje różne obserwowalne:
package dvp.rxjava.observables.exemples;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple13 {
public static void main(String[] args) throws InterruptedException {
// proces 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 1, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// proces 2
Process<String> process2 = new Process<>(
new ProcessAction01<String>("process2", 2, i -> String.format("valeur-%s", i)), Schedulers.computation(), null);
// proces 3
Process<Integer> process3 = new Process<>(new ProcessAction01<Integer>("process3", 3, i -> i * 2), null,
Schedulers.computation());
// proces 4
Process<Boolean> process4 = new Process<>(new ProcessAction01<Boolean>("process4", 4, i -> i % 2 == 0), null, null);
// subskrypcje
ProcessUtils.subscribe(1, process1);
ProcessUtils.subscribe(1, process2);
ProcessUtils.subscribe(1, process3);
ProcessUtils.subscribe(1, process4);
}
}
- wiersze 13–15: proces process1 generuje 1 liczbę rzeczywistą w wątku obliczeniowym, która będzie obserwowana w innym wątku obliczeniowym;
- wiersze 17–18: proces process2 generuje 2 ciągi znaków w wątku obliczeniowym, przy czym nie podano informacji o wątku obserwatora. Wyniki pokazują, że obserwacja odbywa się domyślnie w tym samym wątku, w którym wykonywany jest proces;
- wiersze 20–21: proces process3 generuje 3 liczby całkowite w nieokreślonym wątku, które będą obserwowane w wątku obliczeniowym. Wyniki pokazują, że proces jest domyślnie wykonywany w wątku głównym;
- wiersz 23: proces process4 generuje 4 wartości logiczne w nieokreślonym wątku, które będą obserwowane w nieokreślonym wątku. Wyniki pokazują, że zarówno wykonanie procesu, jak i jego obserwacja odbywają się domyślnie w wątku głównym;
Wynik wykonania tego kodu jest następujący:
- proces process1 generuje 1 liczbę rzeczywistą (wiersz 4) w wątku obliczeniowym [RxComputationThreadPool-4], która jest obserwowana w wątku obliczeniowym [RxComputationThreadPool-3] (wiersz 6);
- proces process2 generuje 2 ciągi znaków (wiersze 12, 14) w wątku obliczeniowym [RxComputationThreadPool-5], które są obserwowane w tym samym wątku (wiersze 13, 15);
- proces process3 generuje 3 liczby całkowite (wiersze 21, 23, 25) w wątku głównym, które są obserwowane w wątku obliczeniowym [RxComputationThreadPool-6] (wiersze 22, 24, 28);
- proces process4 generuje 4 wartości logiczne (wiersze 34, 36, 38, 40) w wątku głównym, które są obserwowane w tym samym wątku głównym (wiersze 33, 35, 37, 39);
Zachęcamy czytelnika do prześledzenia powyższego:
- cykl życia obserwowanego procesu i jego wątku;
- cykl życia jego obserwatora i jego wątku;
Duża część atrakcyjności bibliotek Rx wynika właśnie z tej wielowątkowości, którą programista nie musi samodzielnie zarządzać.
7.5. Kombinacje wielu obserwowalnych
7.5.1. Przykład 14: scalanie dwóch obserwowalnych za pomocą [Observable.merge]
Przedstawiamy teraz metody statyczne klasy [Observable], umożliwiające połączenie kilku obserwowalnych w jedną obserwowalną wynikową.
Pierwszym przykładem tego rodzaju będzie następujący:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.ProcessAction01;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.schedulers.Schedulers;
public class Exemple14 {
public static void main(String[] args) throws InterruptedException {
// proces 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// proces 2
Process<String> process2 = new Process<>(
new ProcessAction01<String>("process2", 2, i -> String.format("valeur-%s", i)), Schedulers.computation(), null);
// scalenie
Process<?> process12 = new Process<>("process12",
Observable.merge(process1.getObservable(), process2.getObservable()));
// subskrypcje
ProcessUtils.subscribe(1, process12);
}
}
- wiersze 15–17: proces o nazwie [process1] wygeneruje 3 liczby rzeczywiste w wątku obliczeniowym. Będzie on również obserwowany w wątku obliczeniowym;
- wiersze 19–20: proces o nazwie [process2] wygeneruje 2 ciągi znaków w wątku obliczeniowym. Wątek obserwacyjny nie jest narzucony. Jak widzieliśmy wcześniej, w tym przypadku wątek obserwacyjny jest wątkami obliczeniowym;
- wiersz 23: oba procesy są scalane, tzn. tworzona jest obserwowalna, której elementy pochodzą jednocześnie z obu procesów. W tym celu wykorzystuje się metodę statyczną [Observable.merge]:
 |
W przeciwieństwie do tego, co mogłoby sugerować powyższy schemat, podczas scalania elementy strumienia 1 mogą wstawiać się pomiędzy elementy strumienia 2. Pokazują to wyniki wykonania:
- wiersz 3: proces [process1] jest wykonywany na wątku obliczeniowym [RxComputationThreadPool-4];
- wiersz 4: proces [process2] jest wykonywany na wątku obliczeniowym [RxComputationThreadPool-5];
- wiersz 9: proces [process12] jest obserwowany na wątku obliczeniowym [RxComputationThreadPool-3]. Nie znam reguły, która doprowadziła do tego wyboru;
- wiersze 9–11: widać, że obserwator obserwuje elementy obu procesów [process1] (wiersz 5) i [process2] (wiersze 6, 7), podczas gdy żaden z nich nie został zakończony (występuje mieszanie);
- proces [process12] kończy działanie (wiersz 17), gdy zakończą się oba procesy process1 i process2;
7.5.2. Przykład 15: łączenie dwóch obserwowalnych za pomocą [Observable.concat]
Przeanalizujmy teraz poniższy kod:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.ProcessAction01;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.schedulers.Schedulers;
public class Exemple15 {
public static void main(String[] args) throws InterruptedException {
// proces 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// proces 2
Process<String> process2 = new Process<>(
new ProcessAction01<String>("process2", 2, i -> String.format("valeur-%s", i)), null, Schedulers.computation());
// łączenie
Process<?> process12 = new Process<>("process12",
Observable.concat(process1.getObservable(), process2.getObservable()));
// subskrypcje
ProcessUtils.subscribe(1, process12);
}
}
- wiersze 15–17: proces o nazwie [process1] wygeneruje 3 liczby rzeczywiste w wątku obliczeniowym. Będzie on również obserwowany w wątku obliczeniowym;
- wiersze 19–20: proces o nazwie [process2] wygeneruje 2 ciągi znaków w wątku nieokreślonym, w tym przypadku domyślnym wątku głównym. Będzie on obserwowany w wątku obliczeniowym;
- wiersz 23: oba procesy są łączone, tzn. tworzona jest obserwowalna, której elementy pochodzą z obu procesów. Nie dochodzi do mieszania wysyłanych wartości. Proces [process12] najpierw wyemituje wszystkie wartości z procesu [process1], a następnie te z procesu [process2]. W tym celu wykorzystuje się metodę statyczną [Observable.concat]:
 |
Wyniki wykonania są następujące:
- wiersze 3–10: proces [process1] jest wykonywany, a proces [process12] wysyła wartości wygenerowane przez [process1];
- wiersz 9: proces [process1] został zakończony;
- wiersze 11–17: proces [process2] jest uruchomiony, a proces [process12] wysyła wartości przekazane przez proces [process2];
W przypadku procesu process2 występuje pewna nieścisłość: nie określono wątku wykonawczego. Można by zatem oczekiwać, że domyślnie będzie to wątek główny. Jednak tak nie jest. Wątkiem wykonawczym okazał się wątek obliczeniowy [RxComputationThreadPool-3] (wiersz 11). Zatem gdy nie określa się wątku wykonawczego ani obserwacyjnego, nie można przewidzieć, który wątek zostanie wybrany.
7.5.3. Przykład 16: łączenie dwóch obserwowalnych wielkości za pomocą [Observable.zip]
Przeanalizujmy teraz następujący kod:
package dvp.rxjava.observables.exemples;
import java.util.Arrays;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.functions.FuncN;
import rx.schedulers.Schedulers;
public class Exemple16 {
public static void main(String[] args) throws InterruptedException {
// proces 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// proces 2
Process<String> process2 = new Process<>(
new ProcessAction01<String>("process2", 2, i -> String.format("valeur-%s", i)), null, null);
// funkcja łącząca oba procesy
FuncN<String> funcn = new FuncN<String>() {
@Override
public String call(Object... args) {
if (args.length == 2) {
return String.format("double=%s, string=%s", args[0], args[1]);
} else {
throw new RuntimeException("la fonction attend 2 paramètres exactement");
}
}
};
// kompresja obu procesów
Process<String> process12 = new Process<>("process12",
Observable.zip(Arrays.asList(process1.getObservable(), process2.getObservable()), funcn));
// subskrypcje
ProcessUtils.subscribe(1, process12);
}
}
- wiersze 16–18: proces o nazwie [process1] wygeneruje 3 liczby rzeczywiste w wątku obliczeniowym. Będzie on również obserwowany w wątku obliczeniowym;
- wiersze 20–21: proces o nazwie [process2] wygeneruje 2 ciągi znaków w wątku nieokreślonym. Wątek obserwacyjny również nie jest określony;
- wiersze 23–32: instancja typu [FuncN<String>] z wykorzystaniem klasy anonimowej. FuncN jest interfejsem funkcjonalnym:
 |
Metoda [FuncN.call] oczekuje tablicy obiektów i zwraca typ R. Funkcja [funcn] zostanie wykorzystana do połączenia procesów process1 i process2 w tej kolejności. W metodzie [FuncN.call]:
- args[0] będzie miało wartością Double;
- args[1] będzie miało wartością String;
W tym przypadku wynikiem metody [funcn.call] będzie ciąg znaków z wiersza 27. Obliczenie tego wyniku nie wymaga znajomości typów argumentów metody call.
Oba procesy łączy się w następujący sposób:
// kompresja obu procesów
Process<String> process12 = new Process<>("process12",
Observable.zip(Arrays.asList(process1.getObservable(), process2.getObservable()), funcn));
Metoda [Observable.zip] działa w następujący sposób:
 |
Widać, że:
- pierwszym argumentem funkcji zip jest Iterable<Observable>. W naszym przykładzie mamy rzeczywisty parametr typu List<Observable> utworzony z naszych dwóch obserwowalnych wielkości;
- drugim argumentem funkcji zip jest typ FuncN. W naszym przykładzie rzeczywistym parametrem jest [funcn];
Wykonanie daje następujące wyniki:
- wiersze 7, 11: proces process12 generuje dwa elementy;
- wiersz 8: dodatkowy element wysłany przez proces process1, który nie ma partnera w procesie process2, nie jest wysyłany przez proces wynikowy process12;
Widać, że proces process2, któremu nie narzucono ani wątku wykonawczego, ani wątku obserwacyjnego, wykorzystał wątek główny do obu tych zadań.
7.5.4. Przykład 17: połączenie dwóch obserwowalnych wielkości za pomocą [Observable.combineLatest]
Przeanalizujmy teraz następujący kod:
package dvp.rxjava.observables.exemples;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.schedulers.Schedulers;
public class Exemple17 {
public static void main(String[] args) throws InterruptedException {
// proces 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// proces 2
Process<Double> process2 = new Process<>(
new ProcessAction01<Double>("process2", 2, i -> new Random().nextInt(200) * 1.4), null,
Schedulers.computation());
// połączenie obu procesów
Process<Double> process12 = new Process<>("process12",
Observable.combineLatest(process1.getObservable(), process2.getObservable(), (d1, d2) -> d1 + d2));
// subskrypcje
ProcessUtils.subscribe(1, process12);
}
}
- wiersze 14–16: proces o nazwie [process1] wygeneruje 3 liczby rzeczywiste w wątku obliczeniowym. Będzie on również obserwowany w wątku obliczeniowym;
- wiersze 18–20: proces o nazwie [process2] wygeneruje 2 liczby rzeczywiste w wątku nieokreślonym. Liczby te będą obserwowane w wątku obliczeniowym;
- wiersz 23: obie obserwowalne są połączone za pomocą następującej metody statycznej [Observable.combineLatest]:
 |
Obserwowalna [combineLatest] działa w następujący sposób: gdy jedna z dwóch obserwowalnych generuje element E1, element ten jest łączony przez [combineFunction] z ostatnim elementem wygenerowanym przez drugą obserwowalną.
Wykonanie tego kodu daje następujący wynik:
- wiersz 5: transmisja process2 (56) jest łączona z ostatnim elementem wysłanym przez process1 (54, wiersz 4) i daje wynik z wiersza 7;
- wiersz 6: transmisja z process1 (51,6) jest łączona z ostatnim elementem przesłanym przez process2 (56, wiersz 5) i daje wynik z wiersza 8;
- wiersz 9: transmisja z process2 (261,8) jest łączona z ostatnim elementem wysłanym przez process1 (51,6, wiersz 6) i daje wynik z wiersza 12;
- wiersz 13: transmisja z process1 (80,39) jest łączona z ostatnim elementem wysłanym przez process2 (261,8, wiersz 9) i daje wynik z wiersza 15;
Mamy tu do czynienia z odmianą obserwowalnej [zip], w której tym razem połączone elementy niekoniecznie są elementami zajmującymi tę samą pozycję w strumieniach. Warto zauważyć, że proces process2, dla którego nie określono wątku wykonania, został tutaj wykonany w wątku głównym (wiersz 2).
7.5.5. Przykład 18: łączenie dwóch obserwowalnych za pomocą [Observable.amb]
Przeanalizujmy teraz następujący kod:
package dvp.rxjava.observables.exemples;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.schedulers.Schedulers;
public class Exemple18 {
public static void main(String[] args) throws InterruptedException {
// proces 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// proces 2
Process<Double> process2 = new Process<>(
new ProcessAction01<Double>("process2", 2, i -> new Random().nextInt(200) * 1.4), null, null);
// połączenie 2 procesów
Process<Double> process12 = new Process<>("process12",
Observable.amb(process1.getObservable(), process2.getObservable()));
// subskrypcje
ProcessUtils.subscribe(1, process12);
}
}
- wiersze 14–16: proces o nazwie [process1] wygeneruje 3 liczby rzeczywiste w wątku obliczeniowym. Będzie on również obserwowany w wątku obliczeniowym;
- wiersze 18–20: proces o nazwie [process2] wygeneruje 2 liczby rzeczywiste w wątku nieokreślonym. Będą one obserwowane w wątku nieokreślonym;
- wiersz 22: obie obserwowalne są połączone za pomocą następującej metody statycznej [Observable.amb]:
 |
Jak pokazuje powyższy schemat, obserwowalna [Observable.amb(Observable o1, Observable o2)] generuje elementy obserwowalnej, która generuje je jako pierwsza. Potwierdzają to wyniki przedstawionego przykładu:
- w wierszu 4 to proces process2 wysyła jako pierwszy;
- wiersze 8, 12: proces process12 generuje wszystkie elementy wygenerowane przez proces process2 (wiersze 4, 11);
7.6. Łańcuch przetwarzania obserwowalnej
7.6.1. Przykład 19: przekształcanie obserwowalnej za pomocą [Observable.map]
W poprzednich przykładach przeanalizowaliśmy różne kombinacje dwóch obserwowalnych w celu uzyskania trzeciego obserwowalnego. Teraz przedstawimy metody statyczne klasy [Observable], które umożliwiają operacje transformacji, filtrowania i agregacji na obserwowalnej. Znajdziemy tu metody analogiczne do tych z klasy [Stream] omówionych w paragrafie 5.
Naszym pierwszym przykładem będzie następujący:
package dvp.rxjava.observables.exemples;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple19 {
public static void main(String[] args) throws InterruptedException {
// proces 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// proces 2
Process<String> process2 = new Process<>("process2",
process1.getObservable().map(d -> String.format("valeur-%s", d)));
// subskrypcje
ProcessUtils.subscribe(1, process2);
}
}
- wiersze 14–16: proces o nazwie process1 wygeneruje 3 liczby rzeczywiste w wątku obliczeniowym. Będzie on również obserwowany w wątku obliczeniowym;
- wiersze 17–18: liczby wygenerowane przez proces process1 zostaną przekształcone w ciągi znaków w procesie process2;
- wiersz 20: obserwujemy process2;
Metoda [Observable.map] z wiersza 18 jest analogiczna do metody [Stream.map] omówionej w paragrafie 5.5:
 |
Wyniki tego przykładu są następujące:
- wiersze 4, 5 i 8: wyniki metody process1. Są to liczby rzeczywiste;
- wiersze 6, 7, 10: emisje process2, które zaobserwowano. Są to ciągi znaków;
7.6.2. Przykład 20: filtrowanie obserwowalnej wielkości za pomocą [Observable.filter]
Przykład będzie wyglądał następująco:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple20 {
public static void main(String[] args) throws InterruptedException {
// proces 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// proces 2
Process<Integer> process2 = new Process<>("process2", process1.getObservable().filter(i -> i % 2 == 0));
// subskrypcje
ProcessUtils.subscribe(1, process2);
}
}
- wiersze 11–12: proces o nazwie process1 będzie generował liczby całkowite od 0 do 2 w wątku obliczeniowym. Będzie on również obserwowany w wątku obliczeniowym;
- wiersz 14: liczby generowane przez proces process1 zostaną przefiltrowane, tak aby w procesie process2 pozostały wyłącznie liczby parzyste;
- wiersz 20: obserwujemy process2;
Metoda [Observable.filter] z wiersza 18 jest analogiczna do metody [Stream.filter] omówionej w paragrafie 5.4:
 |
Wyniki tego przykładu są następujące:
- wiersze 4, 5 i 7: transmisje process1;
- wiersze 6, 9: zaobserwowane emisje process2. Są to elementy process1, które mają parzystą liczbę;
7.6.3. Przykład 21: przekształcenie wielkości obserwowanej za pomocą [Observable.flatMap]
Przykład będzie wyglądał następująco:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.schedulers.Schedulers;
public class Exemple21 {
public static void main(String[] args) throws InterruptedException {
// proces 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// proces 2
Process<Integer> process2 = new Process<>("process2", process1.getObservable().flatMap(i -> {
int value = i * 10;
return Observable.just(value, value + 1, value + 2);
}));
// subskrypcje
ProcessUtils.subscribe(1, process2);
}
}
- wiersze 12–13: proces o nazwie process1 będzie generował liczby całkowite od 0 do 2 w wątku obliczeniowym. Będzie on również obserwowany w wątku obliczeniowym;
- wiersze 15–18: każda liczba n generowana przez proces process1 jest przekształcana w obserwowalną, generującą trzy liczby (10*n, 10*n+1, 10*n+2). Gdyby w wierszu 15 użyto metody [map], metoda process2 generowałaby typ Observable<Integer>, a nie typ Integer. Zastosowana metoda [flatMap] pozwala spłaszczyć (flatten) ten ciąg elementów typu Observable<Integer> do ciągu elementów typu Integer, składającego się z poszczególnych elementów każdego z Observable<Integer>;
- wiersz 20: występuje process2;
Metoda [Observable.flatMap] z wiersza 15 jest analogiczna do metody [Stream.flatMap] omówionej w paragrafie 5.6.12:
 |
Wyniki tego przykładu są następujące:
- wiersze 5–7: trzy transmisje process2 następujące po transmisji z wiersza 4 o numerze process1;
- wiersze 9–11: trzy transmisje process2 w następstwie transmisji z wiersza 8 process1;
- wiersze 14–16: trzy transmisje process2 w następstwie transmisji w wierszu 12 process1;
Poniższy kod pokazuje, jak utworzyć typ Observable<Integer[]> na podstawie process1 i [Exemple21b]:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple21b {
public static void main(String[] args) throws InterruptedException {
// proces 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// proces 2
Process<Integer[]> process2 = new Process<>("process2", process1.getObservable().map(i -> {
int value = i * 10;
return new Integer[] { value, value + 1, value + 2 };
}));
// subskrypcje
ProcessUtils.subscribe(1, process2);
}
}
- wiersz 14: używamy metody [Observable.map];
- wiersz 16: która zwraca typ Integer[];
Wyniki są następujące:
- wiersze 6, 7, 10: widoczne są wyniki metody map;
Wszystkie te transformacje obserwowalnej można łączyć w łańcuch, ponieważ każda transformacja generuje nową obserwowalną. Pokazuje to poniższy przykład [Exemple21c]:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.schedulers.Schedulers;
public class Exemple21c {
public static void main(String[] args) throws InterruptedException {
// proces 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// proces 2
Process<Integer> process2 = new Process<>("process2", process1.getObservable().flatMap(i -> {
int value = i * 10;
return Observable.just(value, value + 1, value + 2);
}).filter(i -> i % 2 == 0));
// subskrypcje
ProcessUtils.subscribe(1, process2);
}
}
- wiersze 15–18: po flatMap następuje filter;
Wyniki wykonania są następujące:
- wiersze 8–13: kod process2 wygenerował jedynie elementy parzyste pochodzące z kodu flatMap;
Metodą zbliżoną do [flatMap] jest metoda [flatMapIterable], zilustrowana poniższym przykładem [Exemple21d]:
package dvp.rxjava.observables.exemples;
import java.util.Arrays;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple21d {
public static void main(String[] args) throws InterruptedException {
// proces 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// proces 2
Process<Integer> process2 = new Process<>("process2", process1.getObservable().flatMapIterable(i -> {
int value = i * 10;
return Arrays.asList(value, value + 1, value + 2);
}).filter(i -> i % 2 == 0));
// subskrypcje
ProcessUtils.subscribe(1, process2);
}
}
W wierszu 16 zamiast metody [flatMap] stosuje się metodę [flatMapIterable]. W tym przypadku funkcja transformacji musi generować typ Iterable<T> (wiersz 18) zamiast typu Observable<T>.
Otrzymujemy takie same wyniki jak poprzednio.
Wróćmy do definicji metody [flatMap]:
 |
Jak widać powyżej, między dwoma zielonymi elementami [1-2] wstawiono niebieski element [3]. Oznacza to, że podczas operacji spłaszczania elementów Observable<T> metoda [flatMap] zachowuje kolejność emisji tych różnych obserwowalnych elementów wewnętrznych. Ilustruje to poniższy przykład [Exemple21e]:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple21e {
public static void main(String[] args) throws InterruptedException {
// proces 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 2, i -> i), Schedulers.computation(),
Schedulers.computation());
// proces 2
Process<Integer> process2 = new Process<>(new ProcessAction01<>("process2", 3, i -> i + 10),
Schedulers.computation(), Schedulers.computation());
// proces 3
Process<Integer> process3 = new Process<>("process3",
process1.getObservable().flatMap(i -> process2.getObservable()));
// subskrypcje
ProcessUtils.subscribe(1, process3);
}
}
- wiersze 11–12: proces process1 generuje liczby całkowite [0,1];
- wiersze 14–15: proces process2 generuje liczby całkowite [10,11,12];
- wiersze 17–18: każdemu elementowi generowanemu przez proces process1 przypisana jest obserwowalna wielkość procesu process2. Oznacza to, że:
- elementowi [0] z procesu 1 zostanie przypisana obserwowalna wielkość generująca [10,11,12];
- to samo dotyczy elementu 1;
Ostatecznie zostanie wyemitowanych 6 liczb [10, 11, 12, 10, 11, 12]. Chcemy sprawdzić, w jakiej kolejności.
Wyniki wykonania są następujące:
Widać, że kolejność generowania przez proces process3 była następująca: [10, 10, 11, 12, 11, 12] (wiersze 11, 12, 14, 17, 19, 22). Doszło zatem do pomieszania elementów wygenerowanych przez proces process2. Można tego uniknąć, stosując metodę [concatMap] zamiast metody [flatMap]. Pokazuje to poniższy kod [Exemple21ef]:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple21ef {
public static void main(String[] args) throws InterruptedException {
// proces 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 2, i -> i), Schedulers.computation(),
Schedulers.computation());
// proces 2
Process<Integer> process2 = new Process<>(new ProcessAction01<>("process2", 3, i -> i + 10),
Schedulers.computation(), Schedulers.computation());
// proces 3
Process<Integer> process3 = new Process<>("process3",
process1.getObservable().concatMap(i -> process2.getObservable()));
// subskrypcje
ProcessUtils.subscribe(1, process3);
}
}
W wierszu 18 zastąpiono [flatMap] na [concatMap]. Wyniki wykonania są następujące:
Widać, że kolejność generowania przez proces process3 była następująca: [10, 11, 12, 10, 11, 12] (wiersze 12–14, 17, 19, 22). Elementy wygenerowane przez proces process2 nie zostały zmieszane.
Inną odmianą metody [map] jest metoda [switchMap]:
 |
Powyżej, z obserwowalnego [1] powstają 3 kolejne obserwowalne [2] składające się z 2 elementów, które są następnie spłaszczane tak jak w [flatMap] i [3]. Można zauważyć, że wynik ma 5 elementów, a nie 6. Wynika to z faktu, że zanim druga obserwowalna wyemitowała swój element nr 2 [6], trzecia obserwowalna wyemitowała już swój pierwszy element [5], co powoduje, że druga obserwowalna zostaje pominięta. W związku z tym elementu [6] nie ma w obserwowalnej wartości wynikowej [3].
Aby zilustrować [switchMap], posłużymy się następującym przykładem [Exemple21eg]:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple21eg {
public static void main(String[] args) throws InterruptedException {
// proces 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 2, i -> i), Schedulers.computation(),
Schedulers.computation());
// proces 2
Process<Integer> process2 = new Process<>(new ProcessAction01<>("process2", 3, i -> i + 10),
Schedulers.computation(), Schedulers.computation());
// proces 3
Process<Integer> process3 = new Process<>("process3",
process1.getObservable().switchMap(i -> process2.getObservable()));
// subskrypcje
ProcessUtils.subscribe(1, process3);
}
}
Wykonanie tego przykładu daje następujące wyniki:
- process1 generuje 2 elementy, które dają początek 2 obserwowalnym process2 składającym się z 3 elementów;
- wiersz 14: obserwator odbiera element nr 0 wysłany przez pierwszą obserwowalną process2 z wiersza 6;
- wiersz 15: obserwator odbiera element nr 0 wysłany przez drugą obserwowalną wielkość process2 z wiersza 13. Nie wiadomo, dlaczego obserwator nie otrzymał wcześniej elementów 1 i 2 wysłanych przez pierwszy obiekt obserwowalny process2 w wierszach 7 i 8. W każdym razie pierwszy obiekt obserwowalny process2 zostaje pominięty;
- w rezultacie obserwator widzi tylko 4 elementy (wiersze 14, 15, 17, 20) zamiast 6, które zostały wysłane;
7.6.4. Przykłady-22: inne metody klasy [Observable]
Klasa [Observable] przejmuje wiele metod z klasy [Stream], działających w podobny sposób. Oto kilka z nich. Podajemy jedynie kod i jego wyniki.
[Exemple22a - take=limit]
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
public class Exemple22a {
public static void main(String[] args) throws InterruptedException {
// proces
Process<Integer> process = new Process<>("process", Observable.range(1, 10).take(3));
// subskrypcje
ProcessUtils.subscribe(1, process);
}
}
wyniki
[Exemple22b - takeLast]
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
public class Exemple22b {
public static void main(String[] args) throws InterruptedException {
// proces
Process<Integer> process = new Process<>("process", Observable.range(1, 10).takeLast(2));
// subskrypcje
ProcessUtils.subscribe(1, process);
}
}
wyniki
[Exemple22c - skip]
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
public class Exemple22c {
public static void main(String[] args) throws InterruptedException {
// procesy
Process<Integer> process = new Process<>("process", Observable.range(1, 10).skip(5).take(2));
// subskrypcje
ProcessUtils.subscribe(1, process);
}
}
wyniki
[Exemple22d - reduce]
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
public class Exemple22d {
public static void main(String[] args) throws InterruptedException {
// procesy
Process<Integer> process = new Process<>("process", Observable.range(1, 10).reduce(0, (i, a) -> i + a));
// subskrypcje
ProcessUtils.subscribe(1, process);
}
}
- wiersz 10: oblicza sumę elementów obserwowalnej. Wynikiem jest obserwowalna, która generuje tę sumę;
wyniki
[Exemple22e - all]
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
public class Exemple22e {
public static void main(String[] args) throws InterruptedException {
// procesy
Process<Boolean> process = new Process<>("process", Observable.range(1, 10).all(i -> i > 10));
// subskrypcje
ProcessUtils.subscribe(1, process);
}
}
- wiersz 10: zwraca Observable<Boolean>, który generuje element true, jeśli predykat metody [all] jest prawdziwy dla wszystkich elementów, w przeciwnym razie zwraca false;
wyniki
[Exemple22f - count]
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
public class Exemple22f {
public static void main(String[] args) throws InterruptedException {
// procesy
Process<Integer> process = new Process<>("process", Observable.range(1, 10).count());
// subskrypcje
ProcessUtils.subscribe(1, process);
}
}
- wiersz 10: [Observable.count] tworzy obserwowalną o jednym elemencie, która jest sumą obserwowanych elementów;
wyniki
[Exemple22g - distinct]
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
public class Exemple22g {
public static void main(String[] args) throws InterruptedException {
// procesy
Process<Integer> process = new Process<>("process", Observable.just(1, 2, 1, 3).distinct());
// subskrypcje
ProcessUtils.subscribe(1, process);
}
}
wyniki
[Exemple22h - groupBy, asObservable]
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.observables.GroupedObservable;
public class Exemple22h {
public static void main(String[] args) throws InterruptedException {
// procesy
Observable<GroupedObservable<Boolean, Integer>> obs = Observable.range(1, 10).groupBy(i -> i % 2 == 0);
Process<Integer> process = new Process<>("process", obs.concatMap(g -> g.asObservable()));
// subskrypcje
ProcessUtils.subscribe(1, process);
}
}
- wiersz 11: metoda [groupBy] grupuje 10 wygenerowanych elementów w 2 grupy: liczby parzyste i nieparzyste. Wynikiem jest typ Observable<GroupedObservable<Boolean, Integer>>, czyli obserwowalna, której elementy są typu GroupedObservable<Boolean, Integer>, gdzie Boolean jest typem klucza grupy (w tym przypadku false, true), który jest również typem wyniku wyrażenia lambda przekazanego jako parametr do metody [groupBy], a Integer jest typem elementów grupy;
- wiersz 12: typ GroupedObservable posiada metodę [asObservable], która pozwala utworzyć obserwowalną na podstawie tego typu. Będziemy więc mieli dwa typy Observable<Integer>, jeden dla liczb parzystych, a drugi dla liczb nieparzystych. Z tych dwóch obserwowalnych metoda [concatMap] utworzy jedną;
wyniki
[Exemple22i - timestamp]
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
import rx.schedulers.Timestamped;
public class Exemple22i {
public static void main(String[] args) throws InterruptedException {
// proces 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// proces 2
Process<Timestamped<Integer>> process2 = new Process<>("process2", process1.getObservable().timestamp());
// subskrypcje
ProcessUtils.subscribe(1, process2);
}
}
- w wierszu 15 metoda [timestamp] przypisuje godzinę do każdego elementu przetwarzanej obserwowalnej;
wyniki
W tym przykładzie trudno jest określić, co oznacza informacja timestamp:
- wiersze 4–5: widać, że element 1 z process1 został wysłany 139 ms po elemencie 0;
- wiersze 6 i 7: widać, że element 1 z process2 został zarejestrowany 234 ms po elemencie 0;
- wiersze 5, 8: widać, że element 2 z process1 został wysłany 33 ms po elemencie 1;
- wiersze 7 i 10: widać, że element 2 z process2 został zaobserwowany 37 ms po elemencie 1;
Te opóźnienia wynikają z faktu, że wątki obserwacji i wykonywania obserwowalnych nie są takie same. Jeśli zastąpimy wiersze 12–13 następującymi wierszami (Przykład22j):
// proces 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
null);
- wiersze 2–3: nie narzuca się wątku obserwacji. Wiadomo, że w takim przypadku wielkość obserwowalna jest obserwowana tam, gdzie jest wykonywana;
Daje to następujące wyniki:
- wiersze 4 i 6: proces process1 wysyła swój element nr 1 587 ms po wysłaniu elementu nr 0;
- wiersze 5 i 7: obserwator obserwuje te dwa elementy z różnicą 586 ms;
- wiersze 6 i 8: proces process1 wysyła swój element nr 2 396 ms po wysłaniu elementu nr 1;
- wiersze 7 i 9: obserwator rejestruje te dwa elementy w odstępie 396 ms;
W tym przypadku wartości procesu timestamp są spójne: faktycznie odzwierciedlają datę wysłania elementu.
7.7. Harmonogramy
7.7.1. Przykład 23: harmonogram [Schedulers.computation]
Przechodzimy teraz do omówienia harmonogramów wykonania. Analiza będzie dotyczyła wątku wykonawczego.
Temat harmonogramów jest nieco niejasny. Różne harmonogramy zostały przedstawione w tym pytaniu na stronie StackOverflow [http://stackoverflow.com/questions/31276164/rxjava-schedulers-use-cases]:
 |
Postaramy się zilustrować zastosowanie tych różnych harmonogramów na przykładach. Pierwszy z nich ilustruje harmonogram [Schedulers.computation]:
package dvp.rxjava.observables.exemples;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple23 {
public static void main(String[] args) throws InterruptedException {
// procesy
@SuppressWarnings("unchecked")
Process<Double> processes[] = new Process[10];
for (int i = 0; i < processes.length; i++) {
processes[i] = new Process<>(
new ProcessAction01<Double>(String.format("process%s", i), 1, value -> new Random().nextInt(100) * 1.2),
Schedulers.computation(), null);
}
// subskrypcje
ProcessUtils.subscribe(1, processes);
}
}
- wiersze 14–19: tworzymy tablicę zawierającą 10 procesów uruchamianych w jednym wątku obliczeniowym;
- wiersz 17: każdy proces generuje losową liczbę rzeczywistą;
- wiersz 21: subskrybujemy wszystkie te procesy;
Wyniki są następujące:
- wiersze 2–10: pierwsze 8 procesów uruchamia się na 8 różnych wątkach (używana maszyna ma 8 rdzeni). Można zauważyć, że wszystkie uruchamiają się mniej więcej w tym samym momencie;
- wiersze 17–19: 3 procesy kończą działanie, zwalniając w ten sposób 3 wątki;
- wiersze 23–24: dwa ostatnie procesy mogą się wówczas uruchomić, zajmując 2 z uwolnionych wątków;
Należy zatem zapamiętać, że harmonogram [Schedulers.computation] udostępnia pulę n wątków, gdzie n to liczba rdzeni komputera. Wątki są wykonywane równolegle na tych rdzeniach.
7.7.2. Przykład 24: harmonogram [Schedulers.io]
Uruchamiamy poprzedni kod z wykorzystaniem harmonogramu [Schedulers.io]:
package dvp.rxjava.observables.exemples;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple24 {
public static void main(String[] args) throws InterruptedException {
// procesy
@SuppressWarnings("unchecked")
Process<Double> processes[] = new Process[10];
for (int i = 0; i < processes.length; i++) {
processes[i] = new Process<>(
new ProcessAction01<Double>(String.format("process%s", i), 1, value -> new Random().nextInt(100) * 1.2),
Schedulers.io(), null);
}
// subskrypcje
ProcessUtils.subscribe(1, processes);
}
}
- wiersz 18: procesy są wykonywane z wykorzystaniem wątków harmonogramu [Schedulers.io];
Daje to następujące wyniki:
- wiersze 2–10: wszystkie 10 procesów uruchamia się w oddzielnych wątkach. W przeciwieństwie do poprzedniego przypadku udało się uruchomić wszystkie procesy. Można zauważyć, że uruchomienie tych procesów trwało 6 ms, podczas gdy wcześniej zajmowało to 1 ms;
- wiersze 13–18: obserwowalne generują dane kolejno, a nie niemal równolegle, jak miało to miejsce wcześniej;
Jaka jest różnica między harmonogramami [Schedulers.io] i [Schedulers.computation]? Odpowiedź można znaleźć w URL [http://stackoverflow.com/questions/31276164/rxjava-schedulers-use-cases]:
 |
7.7.3. Przykład 25: harmonogram [Schedulers.newThread]
Uruchamiamy powyższy kod za pomocą harmonogramu [Schedulers.newThread]:
package dvp.rxjava.observables.exemples;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple25 {
public static void main(String[] args) throws InterruptedException {
// procesy
@SuppressWarnings("unchecked")
Process<Double> processes[] = new Process[10];
for (int i = 0; i < processes.length; i++) {
processes[i] = new Process<>(
new ProcessAction01<Double>(String.format("process%s", i), 1, value -> new Random().nextInt(100) * 1.2),
Schedulers.newThread(), null);
}
// subskrypcje
ProcessUtils.subscribe(1, processes);
}
}
Uzyskane wyniki są takie same jak w przypadku harmonogramu [Schedulers.io]:
W sekcji dotyczącej harmonogramów URL i [http://stackoverflow.com/questions/33415881/retrofit-with-rxjava-schedulers-newthread-vs-schedulers-io] wyjaśniono, że harmonogram [Schedulers.io] udostępnia pulę wątków, czego nie robi harmonogram [Schedulers.newThread]. Pula wątków automatycznie tworzy n wątków. Przydziela je procesom, które ich potrzebują. Gdy procesy te zakończą działanie, ich wątki nie są usuwane, lecz wracają do puli i mogą zostać ponownie wykorzystane przez inny proces. Jest to bardziej ekonomiczne niż ciągłe tworzenie i usuwanie wątków. Można zatem uznać, że lepiej jest używać harmonogramu [Schedulers.io].
7.7.4. Przykład 26: harmonogramy [Schedulers.immediate, Schedulers.trampoline]
Wróćmy do wyjaśnienia dotyczącego tych dwóch harmonogramów:
|
Wyjaśnienie jest dość proste do zrozumienia, ale kiedy chcemy je zilustrować, okazuje się, że tak naprawdę go nie zrozumieliśmy. To właśnie książka [Learning Reactive Programming With Java 8] pozwoliła mi stworzyć przykład, który nawiązuje do przykładu znalezionego w tej książce, ale go upraszcza. Oto on:
package dvp.rxjava.observables.exemples;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.function.Consumer;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Scheduler;
import rx.Scheduler.Worker;
import rx.functions.Action0;
import rx.schedulers.Schedulers;
public class Exemple26 {
public static void main(String[] args) throws InterruptedException {
// harmonogram
Scheduler scheduler = Schedulers.immediate();
// pracownik tego harmonogramu
Worker worker = scheduler.createWorker();
// typ Action0 do wykonania na procesie roboczym
Action0 action02 = new Action0() {
@Override
public void call() {
// log akcji02
ProcessUtils.showInfos.accept("action02");
}
};
// typ Action0 do wykonania na procesie roboczym
Action0 action01 = new Action0() {
@Override
public void call() {
// programujemy nową akcję na tym samym procesie roboczym
worker.schedule(action02);
// log akcji 01
ProcessUtils.showInfos.accept("action01");
}
};
// akcja01 została zaplanowana na workerze
worker.schedule(action01);
}
// wyświetlenia
static Consumer<String> showInfos = message -> System.out.printf("%s ------Thread[%s] ---- Time[%s]%n", message,
Thread.currentThread().getName(), new SimpleDateFormat("ss:SSS").format(new Date()));
}
- wiersz 17: harmonogram. Będzie to albo [Schedulers.immediate], jak tutaj, albo później [Schedulers.trampoline];
- wiersz 19: można uruchamiać działania typu Action0 (wiersze 21, 20) na procesach roboczych harmonogramu. Metoda [Scheduler.createWorker] pozwala utworzyć proces roboczy. Metoda [Worker.schedule(Action0)] pozwala zlecić wykonanie typu Action0 przez proces roboczy;
- wiersze 21–27: pierwsza akcja o nazwie [action02], która zostanie wykonana (wiersz 40) przez proces roboczy z wiersza 19;
- wiersze 30–38: druga akcja o nazwie [action01]. Jej cechą charakterystyczną jest to, że powoduje ona wykonanie akcji action02 na tym samym workerze, co ona sama (wiersz 34). Właśnie na tym polega różnica między [Schedulers.immediate] a [Schedulers.trampoline]:
- jeśli harmonogramem jest [Schedulers.immediate], to w wierszu 34 akcja action02 zostanie wykonana natychmiast (stąd nazwa harmonogramu), a bieżąca akcja action01 zostanie przerwana. Wówczas pojawi się komunikat z wiersza 25. Po zakończeniu akcji action02 akcja action01 zostanie wznowiona i pojawi się komunikat z wiersza 36;
- jeśli harmonogramem jest [Schedulers.trampoline], to w wierszu 34 akcja action02 zostanie zawieszona. Zostanie ona wykonana dopiero po zakończeniu bieżącego zadania action01. Wówczas pojawi się komunikat z wiersza 36. Po zakończeniu akcji action01 zostanie wykonana akcja action02 i zobaczymy komunikat z wiersza 25;
Wykonanie powyższego kodu daje następujące wyniki:
Jeśli w wierszu 17 użyjemy harmonogramu [Schedulers.trampoline], otrzymamy odwrotne wyniki:
Trudno jednak dostrzec związek z obserwowalnymi. Nie znalazłem przekonującego przykładu, który pokazałby korzyści z uruchomienia obserwowalnej na jednym z tych dwóch wątków. Oto jednak jeden przykład, który jednak nie wydaje mi się wcale naturalny:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.Scheduler.Worker;
import rx.functions.Action1;
import rx.schedulers.Schedulers;
public class Exemple27 {
public static void main(String[] args) throws InterruptedException {
// Moduł roboczy
Worker worker = Schedulers.immediate().createWorker();
// Worker worker = Schedulers.trampoline().createWorker();
// obserwowalny 1 na workerze
worker.schedule(() -> Observable.range(1, 2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer i) {
ProcessUtils.showInfos.accept(String.valueOf(i));
// obserwowalna 2 na tym samym workerze
worker.schedule(() -> Observable.range(100, 2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer i) {
ProcessUtils.showInfos.accept(String.valueOf(i));
}
}));
}
}));
}
}
- wiersze 13–14: tworzymy proces roboczy na podstawie jednego z dwóch harmonogramów [Schedulers.immediate] i [Schedulers.trampoline];
- wiersz 16: na tym workerze zaplanowano pierwszą obserwowalną obs1, która ma generować liczby [1,2]
- wiersz 22: za każdym razem, gdy zaobserwowany zostanie element tej obserwowalnej wartości obs1, na tym samym workerze uruchamiana jest obserwacja drugiej obserwowalnej wartości obs2 w celu generowania liczb [100,101];
Dzięki harmonogramowi [Schedulers.immediate] uzyskuje się następujące wyniki:
Natomiast przy użyciu harmonogramu [Schedulers.trampoline] uzyskuje się następujące wyniki:
7.8. Conclusion
Wciąż pozostaje wiele do zrobienia. Aby pogłębić wiedzę na temat biblioteki RxJava, zachęcamy czytelnika do kontynuowania nauki, korzystając z materiałów referencyjnych podanych na początku niniejszego dokumentu. Niemniej jednak dysponujemy już podstawami do wykorzystania RxJava w środowiskach Swing i Android. Właśnie to zamierzamy teraz pokazać.