7. Die Bibliothek RxJava
Die Bibliothek RxJava basiert auf folgendem Konzept: Ein Strom von Elementen vom Typ T Observable<T> wird von einem oder mehreren Abonnenten (Subscriber<T>) beobachtet. Die Bibliothek RxJava ermöglicht es, dass der Observable<T>-Stream in einem Thread T1 und sein Beobachter Subscriber<T> in einem Thread T2 ausgeführt werden, ohne dass sich der Entwicklersich um die Verwaltung des Lebenszyklus dieser Threads und um naturgemäß schwierige Probleme wie den Datenaustausch zwischen Threads und deren Synchronisation zur Ausführung einer übergreifenden Aufgabe kümmern muss. Sie erleichtert somit die asynchrone Programmierung.
Ein Observable<T>-Stream erzeugt Elemente vom Typ T, die nach und nach beobachtet werden können, sobald sie erzeugt werden. Befinden sich der Beobachter und das Observable (wobei der Typ Observable<T> hier im weiteren Sinne verwendet wird) im selben Thread, so kann das Observable das Element (i+1) erst dann erzeugen, wenn der Beobachter das Element i verarbeitet hat. Es gibt nur wenige Fälle, in denen diese Architektur von Interesse ist. Befinden sich der Beobachter und das Observable nicht im selben Thread, verhalten sich das Observable und sein Beobachter unabhängig voneinander: Das Observable erzeugt Elemente in seinem eigenen Tempo und der Beobachter verarbeitet sie in seinem eigenen Tempo. Genau darin liegt der Vorteil der Bibliothek. Bislang haben wir immer von einem Beobachter gesprochen. Tatsächlich kann ein Observable jedoch eine beliebige Anzahl von Beobachtern haben.
Die Bibliothek RxJava eignet sich besonders gut für die in Abschnitt 2 der Einführung beschriebene Architektur, die wir hier noch einmal zusammenfassen:
- In [1] stellt eine Dienstschicht Dienste bereit, von denen einige lange dauern (z. B. Netzwerkanfragen);
- diese Service-Schicht wird von einer grafischen Benutzeroberfläche [1] (Swing, Android, JavaFx) aufgerufen. Wird die Service-Schicht im selben Thread ausgeführt wie die Methode [swing], die sie nutzt, friert die grafische Benutzeroberfläche ein (reagiert nicht), während auf das Ergebnis des Dienstes gewartet wird;
- In [2] ermöglicht eine mit RxJava implementierte dünne Anpassungsschicht, der grafischen Schicht eine asynchrone Implementierung desselben Dienstes bereitzustellen: Dieser kann in einem anderen Thread ausgeführt werden als die Methode der grafischen Schicht, die ihn aufruft. In diesem Fall bleibt die grafische Benutzeroberfläche [3] reaktionsfähig: Der Benutzer kann weiterhin mit ihr interagieren, beispielsweise parallel zur ersten Anfrage eine neue Netzwerkanfrage auslösen, und vor allem kann man ihm die Möglichkeit bieten, zu langwierige Verarbeitungsvorgänge abzubrechen – was unmöglich wäre, wenn die grafische Benutzeroberfläche eingefroren wäre;
- Der Aufruf [4] ist synchron, während der Aufruf [5-6] asynchron ist;
In dieser Architektur stellt die Schicht [2] Dienste bereit, die Typen vom Typ **Observable<T>** zurückgeben, die von den Methoden der Grafikschicht [3] abonniert werden können. Ein Dienst der Schicht [2] liefert dann seine Ergebnisse nacheinander aus, und die Schicht [3] kann auf jedes einzelne davon reagieren, indem sie beispielsweise eine oder mehrere Komponenten der grafischen Benutzeroberfläche aktualisiert.
Die Klasse Observable<T> verfügt über mehrere Dutzend Methoden. Dies ist eine der Herausforderungen der Bibliothek: Sie ist sehr umfangreich, und es ist schwierig, alle ihre Möglichkeiten zu erfassen. Wir werden einige davon vorstellen. Die Beherrschung der übrigen Methoden wird sich dann mit der Zeit einstellen.
7.1. Observables erstellen und abonnieren
7.1.1. Beispiel-01: Die Methode [Observable.from]
 |
Betrachten wir den folgenden Code:
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) {
// Beobachtbare Ganzzahlen
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");
}
});
}
}
- Zeile 12: Aus einer Liste von Ganzzahlen wird ein Typ „Observable<Integer>“ erstellt.
Die Klasse Observable<T> ist ein Strom von Elementen vom Typ T, die beobachtet werden können – vorzugsweise asynchron, aber nicht zwingend –, sobald sie erzeugt werden. Ihre Definition lautet wie folgt:
 |
Wie bereits erwähnt, verfügt die Klasse Observable<T> über mehrere Dutzend Methoden. Einige davon ähneln denen der in Abschnitt 5 behandelten Klasse Stream<T>. Die Dokumentation zu RxJava enthält „Marble-Diagramme“ ([2]), die die Funktionsweise dieser Methoden veranschaulichen:
- Zeile 3 veranschaulicht die Emissionen der Observable im Zeitverlauf;
- die Methode [4] wird auf die vom Observablen emittierten Elemente angewendet. Sie erzeugt im Allgemeinen ein neues Observabel;
- Zeile 5 zeigt die erhaltene neue Observable;
Die Methode [Observable.from] hat folgende Signatur:
 |
Die statische Methode [Observable.from] ermöglicht es, aus einer Sammlung von Elementen vom Typ T ein Observable<T> zu erstellen. Dies ist eine sehr einfache Möglichkeit, mit Observables zu beginnen. Die Zeile:
Observable<Integer> obs1 = Observable.from(Arrays.asList(1, 2, 3));
gibt daher drei Elemente aus. Sie gibt diese jedoch nicht sofort aus. Sie gibt sie jedes Mal vollständig aus, wenn sich ein Beobachter anmeldet. Dies wird als „kaltes Observable“ bezeichnet. Das Observable gibt seine Elemente für jeden neuen Abonnenten erneut aus.
Man kann die vorstehende Anweisung als eine Konfigurationsaktion des Observables betrachten. Dieses wird einmal konfiguriert und n-mal ausgeführt, wenn sich n Abonnenten anmelden.
Wie abonniert man?
Eine Möglichkeit hierfür ist die Verwendung der Methode [Observable.subscribe], deren hier verwendete Definition wie folgt lautet:
 |
- Der erste Parameter [Action1<T> onNext] (siehe Abschnitt 6.2) der Methode ist die Methode, die ausgeführt werden soll, wenn das Observable ein neues Element T ausgibt;
- Der zweite Parameter [Action1<Throwable> onError] der Methode ist die Methode, die ausgeführt werden soll, wenn das Observable eine Ausnahme auslöst;
- Der dritte Parameter [Action0 onComplete] (siehe Abschnitt 6.1) der Methode ist die Methode, die ausgeführt werden soll, wenn das Observable eine Ausnahme auslöst;
- Die Methode gibt einen Typ [Subscription] zurück;
Der Typ [Subscription] stellt ein Abonnement des Observables dar. Seine Definition lautet wie folgt:
 |
Der Vorteil dieser Schnittstelle [1] liegt in ihrer Methode [2], mit der ein Abonnement gekündigt werden kann.
In unserem Beispiel lautet der Code für das Abonnement des Observables wie folgt:
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");
}
});
- Zeile 1: Das Ergebnis vom Typ [Subscription] wird ignoriert;
- Zeilen 1–15: Die drei Parameter sind Instanzen anonymer Klassen. Wir werden außerdem Lambda-Ausdrücke verwenden. Der Vorteil anonymer Klassen besteht darin, dass die von der einzigen Methode dieser Klassen erwarteten Datentypen klar erkennbar sind;
- Zeilen 2–5: Implementierung des ersten Parameters vom Typ [Action1<Integer>];
- Zeilen 6–10: Implementierung des zweiten Parameters vom Typ [Action1<Throwable>];
- Zeilen 11–15: Implementierung des dritten Parameters vom Typ [Action0];
Der gesamte Code lautet wie folgt:
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) {
// Ganzzahl-Observables
Observable<Integer> obs1 = Observable.from(Arrays.asList(1, 2, 3));
// Abonnement
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");
}
});
}
}
Das Observable in Zeile 12 beginnt mit der Ausgabe seiner drei Elemente, sobald die Methode [subscribe] in Zeile 14 aufgerufen wird. Ab diesem Zeitpunkt:
- werden bei jedem emittierten Element die Zeilen 15–18 ausgeführt.
- sind die 3 Elemente ausgegeben, werden die Zeilen 24–29 ausgeführt;
- die Zeilen 19–24 werden niemals ausgeführt, da das Observable hier keine Ausnahme auslöst;
Standardmäßig werden das Observable und der Observer im selben Thread ausgeführt. Es gibt einige vordefinierte Observables, die in einem anderen Thread als dem Hauptthread (hier dem Thread der Methode main) ausgeführt werden, aber für die meisten von ihnen trifft dies nicht zu. Hier spielt sich also alles im Thread der Methode „[main]“ ab:
- Das Observable gibt das Element 1 aus;
- die Zeilen 15–18 werden ausgeführt und zeigen dieses Element an;
- das Observable gibt das Element 2 aus;
- die Zeilen 15–18 werden ausgeführt und zeigen dieses Element an;
- das Observable gibt das Element 3 aus;
- die Zeilen 15–18 werden ausgeführt und zeigen dieses Element an;
- Das Observable sendet die Benachrichtigung „[completed]“;
- die Zeilen 24–29 werden ausgeführt;
Dies zeigen die erzielten Ergebnisse:
Die Klasse [Exemple02] übernimmt [Exemple01], wobei diesmal Lambda-Funktionen als Parameter der Methode [Observable.subscribe] verwendet werden:
package dvp.rxjava.observables;
import java.util.Arrays;
import rx.Observable;
public class Exemple02 {
public static void main(String[] args) {
// Ganzzahl-Observables
Observable<Integer> obs1 = Observable.from(Arrays.asList(1, 2, 3));
// Abonnement
obs1.subscribe(
(integer) -> System.out.printf("next : %s%n", integer),
(th) -> System.out.println(th),
() -> System.out.println("completed"));
}
}
7.1.2. Beispiel-03: Die Observer-Klasse
 |
Die Methode [Observable.subscribe], mit der man ein Observable abonnieren kann, verfügt über verschiedene Versionen, darunter die folgende:
package dvp.rxjava.observables;
import java.util.Arrays;
import rx.Observable;
import rx.Observer;
public class Exemple03 {
public static void main(String[] args) {
// Beobachtbare Ganzzahlen
Observable<Integer> obs1 = Observable.from(Arrays.asList(1, 2, 3));
// Abonnement
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);
}
});
};
}
Zeile 13: Anstatt drei Parameter an die Methode [subscribe] zu übergeben, wird ihr der folgende Typ [Observer] übergeben:
 |
Der Typ [Observer] ist eine Schnittstelle mit drei Methoden:
- [onNext(T t)], die jedes Mal aufgerufen wird, wenn das Observable ein Element t ausgibt;
- [onError(Throwable th)], die aufgerufen wird, wenn das Observable eine Ausnahme th auslöst;
- [onCompleted], die aufgerufen wird, wenn das Observable signalisiert, dass es die Ausgabe beendet hat;
Die Funktionsweise des Codes entspricht der zuvor erläuterten. Man erhält folgende Ergebnisse:
7.1.3. Beispiel-04: Die Methode [Observable.create]
 |
Die statische Methode Observable.create ist wie folgt definiert:
 |
- Die Methode [create] gibt einen Typ „Observable<T>“ zurück;
- Der Parameter der Methode [create] ist eine Funktion vom Typ [Observable.OnSubscribe<T>], die wie folgt definiert ist:
 |
Der Typ [Observable.OnSubscribe<T>] ist eine funktionale Schnittstelle, die ihrerseits die funktionale Schnittstelle [Action1<Subscriber<? super T>>] erweitert. Die Methode [call] dieser Schnittstelle erwartet einen Typ [Subscriber] (Abonnent, Beobachter), der wie folgt definiert ist:
 |
In [1] ist zu sehen, dass die Klasse [Subscriber<T>] die in Abschnitt 7.1.2 vorgestellte Schnittstelle [Observer<T>] implementiert.
Letztendlich erwartet die Methode [<T> Observable.create]:
- erwartet als Parameter eine Instanz vom Typ [Observable.OnSubscribe<T>] mit der einzigen Methode folgender Signatur: void call(Subscriber<T> s). Der Typ [Subscriber<T>] erweitert den Typ [Observer<T>] und verfügt daher über die Methoden onNext, onError, onCompleted;
- gibt einen Typ „Observable<T>“ zurück;
Die Methode [<T> Observable.create] gibt ein konfiguriertes Observable zurück. Bisher wurden noch keine Elemente ausgegeben. Wenn ein Abonnent [Subscriber<T> s] dieses Observable abonniert, wird die Methode [void call(s)] der Funktion aufgerufen, die als Parameter an die Methode [<T> Observable.create] übergeben wurde. Ihre Aufgabe besteht darin, Elemente t vom Typ T auszugeben und bei jeder Ausgabe die Methode [s.onNext(t)] des Beobachters aufzurufen. Sobald diese beendet ist, muss die Methode [s.onCompleted(t)] des Beobachters aufgerufen werden und die Methode [call] beendet werden. Wenn in der Methode [call] eine Ausnahme th auftritt, muss die Methode [s.onError(th)] des Beobachters aufgerufen werden und die Methode [call] beendet werden;
Um diese komplexe Funktionsweise zu veranschaulichen, verwenden wir den folgenden Code [Exemple04]:
package dvp.rxjava.observables;
import rx.Observable;
import rx.Subscriber;
import java.util.Random;
public class Exemple04 {
public static void main(String[] args) {
// Beobachtbare Konfiguration von reellen Zahlen
Observable<Double> obs1 = Observable.create(new Observable.OnSubscribe<Double>() {
@Override
public void call(Subscriber<? super Double> subscriber) {
for (int i = 0; i < 3; i++) {
// Ausgabe des Elements i
subscriber.onNext(new Random((i + 1)).nextDouble());
}
// Ende der Übertragung
subscriber.onCompleted();
}
});
// Abonnement und somit Emission
obs1.subscribe((d) -> System.out.printf("onNext %s%n", d), (th) -> System.out.printf("onError %s%n", th),
() -> System.out.println("onCompleted"));
}
}
- Zeile 11: Es wird ein Observable erstellt, das Typen vom Typ Double emittiert;
- Zeilen 11–21: Der Parameter der Methode [create] wird mit einer anonymen Klasse instanziiert, die die einzige Methode [call] aus den Zeilen 12–20 enthält. Das in Zeile 11 erstellte Observable ist bereit zum Emittieren, wird jedoch erst dann emittieren, wenn ein Beobachter hinzukommt;
- Zeilen 13–21: Die Methode [call] erhält die Referenz eines Beobachters;
- Zeilen 14–17: Senden von 3 Elementen an den Beobachter;
- Zeile 19: Benachrichtigung des Beobachters über das Ende der Übertragung;
- Zeilen 23–24: Abonnement für die Beobachtungsgröße aus Zeile 11. Die drei Parameter [onNext, onError, onCompleted] der Methode [subscribe] werden durch drei Lambda-Ausdrücke implementiert. Dieses Abonnement erstellt den Abonnenten [Subscriber<Double>], der an die Methode [call] in Zeile 13 übergeben wird. Die Emission von Elementen beginnt dann;
- alles läuft im selben Thread ab: Observable und Observer;
Man erhält folgende Ergebnisse:
Mit der Methode [Observable.create] lässt sich aus jedem beliebigen Phänomen ein Observable erstellen. Diese Methode haben wir in Abschnitt 2 der Einführung verwendet, um eine synchrone Schnittstelle in eine asynchrone Schnittstelle umzuwandeln.
7.1.4. Beispiel-05: Refactoring von [Exemple-04]
 |
Das folgende Beispiel zeigt eine neue Version der statischen Methode [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) {
// Konfiguration einer Beobachtbaren für reelle Zahlen
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++) {
// Warten
try {
Thread.sleep(500 - i * 100);
} catch (InterruptedException e) {
// Fehler
subscriber.onError(e);
}
// Aktion
double value = new Random().nextInt(100) * 1.2;
showInfos(String.format("Observable.call onNext(%s)", value));
subscriber.onNext(value);
}
// beendet
showInfos(String.format("Observable.call onCompleted"));
subscriber.onCompleted();
}
});
// ein Abonnent
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));
}
};
// Abonnement
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()));
}
}
- Zeile 56: Die neue Version der statischen Methode [Observable.subscribe] akzeptiert als Parameter den Typ [Subscriber], den wir im vorigen Absatz vorgestellt haben;
- Zeilen 37–52: der Abonnent (Beobachter). Er implementiert die Schnittstelle Observer mit ihren drei Methoden onNext, onError und onCompleted;
- Zeilen 61–64: Von nun an werden wir uns mit den Threads befassen, in denen das Observable und sein Beobachter ausgeführt werden;
- Zeile 62: der Name des Threads;
- Zeile 63: die aktuelle Zeit in Sekunden und Millisekunden. So können wir im Zeitverlauf verfolgen, wie das Observable Elemente ausgibt und wie diese vom Observer verarbeitet werden;
- Dieser Code hat dieselbe Funktionalität wie der vorherige. Wir haben diesen lediglich umgestaltet;
Die erzielten Ergebnisse lauten wie folgt:
- Zeile 1 der Ergebnisse: Vor Zeile 56 des Codes ist noch nichts passiert. Das Observable wurde lediglich konfiguriert;
- Zeile 2 der Ergebnisse: Zeile 56 des Codes löst den Aufruf der Methode [call] aus Zeile 15 aus. In Zeile 3 wird der Realwert 80,39 an den Beobachter gesendet;
- Zeile 4: Der Beobachter empfängt die gesendete Zahl;
- Zeilen 5–8: Der vorangegangene Vorgang wiederholt sich zweimal;
- Zeile 9: Der Beobachter sendet die Benachrichtigung über das Ende der Übertragung;
- Zeile 10: Der Beobachter empfängt sie;
- Zeile 11: wird durch Zeile 57 des Codes angezeigt;
Man sieht also, dass allein die Abonnementzeile 56 die Anzeige der Zeilen 2–10 der Ergebnisse ausgelöst hat. Wenn man mit der Bibliothek RxJava beginnt, fragt man sich, wie die einzelnen Schritte ineinandergreifen und insbesondere, welche Verbindungen zwischen dem Beobachter und dem Beobachtbaren bestehen. Hier sieht man, dass Zeile 56, die Abonnierung des Beobachtbaren,
- die Ausgabe aller Elemente des Beobachtbaren ausgelöst hat;
- dass das Beobachtbare und der Beobachter im selben Thread ausgeführt werden;
- dass man deshalb folgende Abfolge beobachtet: Emission von Element i, Beobachtung von Element i, Emission von Element (i+1), Beobachtung von Element (i+1), …
Wir erinnern uns, dass der Emitter vor der Emission seiner Elemente gewartet hat:
// Warten
try {
Thread.sleep(500 - i * 100);
} catch (InterruptedException e) {
// Fehler
subscriber.onError(e);
}
wobei i in Zeile 3 die Nummer der Übertragung darstellt (0 <= i < 3). Betrachtet man die Übertragungszeitpunkte der Elemente des Beobachtbaren:
- Zeilen 2, 3: Element 0 wurde etwa 500 ms nach Beginn des Abonnements gesendet;
- Zeilen 3, 5: Element 1 wurde etwa 400 ms nach Element 0 gesendet;
- Zeilen 5, 7: Element 2 wurde etwa 300 ms nach Element 1 gesendet;
7.2. Ausführungsthread, Beobachtungs-Thread
7.2.1. Beispiel-06: Beobachtbares und Beobachter in einem anderen Thread als [main]
 |
Wir überarbeiten das vorherige Beispiel wie folgt: [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) {
// Schrankenwärter
CountDownLatch latch = new CountDownLatch(1);
// Konfiguration einer Beobachtungsgröße aus reellen Zahlen
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++) {
// Warten
try {
Thread.sleep(500 - i * 100);
} catch (InterruptedException e) {
// Fehler
subscriber.onError(e);
}
// Aktion
double value = new Random().nextInt(100) * 1.2;
showInfos(String.format("Observable.call onNext(%s)", value));
subscriber.onNext(value);
}
// Beendet
showInfos(String.format("Observable.call onCompleted"));
subscriber.onCompleted();
}
});
// ein Abonnent
Subscriber<Double> subscriber = new Subscriber<Double>() {
@Override
public void onCompleted() {
showInfos("Subscriber.onCompleted");
// Die Schranke wird heruntergelassen
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));
}
};
// Fortsetzung der Konfiguration des Observables
obs1 = obs1.subscribeOn(Schedulers.computation());
// Zeichnung
showInfos("avant souscription");
obs1.subscribe(subscriber);
// Warten vor der Schranke
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()));
}
}
- Zeile 16: Wir erstellen eine Barriere (Semaphor) mit einem Objekt vom Typ [CountDownLatch]. Dieses Objekt dient dazu, Threads untereinander zu synchronisieren. Es wird hier mit dem Wert 1 initialisiert, den wir als Wert der Barriere (oder des Semaphors) bezeichnen. Ein Thread wartet durch eine Operation auf die Barriere:
latch.await();
Der Thread wird blockiert, wenn der Wert des Semaphors > 0 ist. Ein Thread kann den internen Wert des Semaphors erhöhen oder verringern. In Zeile 48 wird der Wert des Semaphors um 1 verringert.
- Zeile 63: Das Observable ist so konfiguriert, dass es auf einem vom Scheduler bereitgestellten Thread ausgeführt wird: [Schedulers.computation()]. Dieser Scheduler kann so viele Threads bereitstellen, wie es Kerne auf dem Ausführungsrechner gibt. Der Abschnitt zur Beispielanwendung hat die Verwendung anderer Scheduler gezeigt (siehe Abschnitt 2.8);
Das Prinzip des Codes ist wie folgt:
- Die Methode [main] wird im Hauptthread (main) ausgeführt;
- Zeile 66: Löst die Ausgabe von Elementen des Observables aus. Diese werden in einem anderen Thread als dem Hauptthread ausgegeben;
- Zeile 70: Der Hauptthread wird blockiert, da der Wert des „Barrier“ 1 beträgt (siehe Zeile 16). Er kann erst fortfahren, wenn dieser Wert auf 0 wechselt. Dies geschieht in Zeile 48. Der Beobachter senkt die Barriere, sobald er die Benachrichtigung erhält, dass das Observable seine Emissionen beendet hat;
Die Ausführung liefert folgende Ergebnisse:
- Zeile 1: Die Abonnierung wird durchgeführt;
- Zeile 2: Diese löst die Ausführung der Methode [call] im Thread [RxComputationThreadPool-1] aus. Es findet nun eine parallele Ausführung mit zwei Threads statt;
- Zeile 3: Aus ungeklärtem Grund hat der Thread [RxComputationThreadPool-1] die Kontrolle abgegeben. Der Thread [main] übernimmt daraufhin die Kontrolle und wird durch die Sperre (Zeile 70 des Codes) blockiert. Ab diesem Zeitpunkt kann nur noch der Thread [RxComputationThreadPool-1] arbeiten;
- Zeilen 4–11: Man beobachtet das zuvor beschriebene Verhalten zwischen der Observable und ihrem Beobachter, doch spielt sich nun alles im Thread [RxComputationThreadPool-1] ab;
- Zeilen 12–13: Der Beobachter hat die Schranke gesenkt (Zeile 48 des Codes) und der Thread [RxComputationThreadPool-1] wurde beendet. Der Thread [main] übernimmt die Kontrolle und gibt zwei Meldungen aus;
7.2.2. Beispiel-07: Observable und Observer in zwei verschiedenen Threads
|
Wir ändern das vorherige Beispiel wie folgt:
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) {
// Schrankenwärter
CountDownLatch latch = new CountDownLatch(1);
// Konfiguration einer Beobachtbaren aus reellen Zahlen
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++) {
// Warten
try {
Thread.sleep(500 - i * 100);
} catch (InterruptedException e) {
// Fehler
subscriber.onError(e);
}
// Aktion
double value = new Random().nextInt(100) * 1.2;
showInfos(String.format("Observable.call onNext(%s)", value));
subscriber.onNext(value);
}
// Beendet
showInfos(String.format("Observable.call onCompleted"));
subscriber.onCompleted();
}
});
// ein Abonnent
Subscriber<Double> subscriber = new Subscriber<Double>() {
@Override
public void onCompleted() {
showInfos("Subscriber.onCompleted");
// Die Schranke wird heruntergelassen
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));
}
};
// Fortsetzung der Konfiguration des Observables
obs1 = obs1.subscribeOn(Schedulers.computation()).observeOn(Schedulers.computation());
// Zeichnung
showInfos("avant souscription");
obs1.subscribe(subscriber);
// Warten auf das Erreichen der Barriere
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()));
}
}
Der Code ist identisch mit dem des vorherigen Beispiels, mit Ausnahme von Zeile 63:
obs1 = obs1.subscribeOn(Schedulers.computation()).observeOn(Schedulers.computation());
die das Observable (subscribeOn) und den Observer (observeOn) so konfiguriert, dass sie auf einem der vom Scheduler [Schedulers.computation()] bereitgestellten Threads ausgeführt werden.
Die erzielten Ergebnisse lauten wie folgt:
Folgende Punkte sind zu beachten:
- Das Observable wird im Thread [RxComputationThreadPool-4] ausgeführt (Zeilen 3–4, 6, 8–9);
- Der Beobachter wird im Thread [RxComputationThreadPool-3] ausgeführt (Zeilen 5, 7, 10–11);
- dass sie unabhängig voneinander ausgeführt werden. So sendet das Observable in den Zeilen 8–9 zwei Benachrichtigungen (onNext, onCompleted), bevor der Observer die Benachrichtigung [onNext] abruft (Zeile 10);
Die Bibliothek RxJava kümmert sich um die Datenübergabe (die Emissionen) vom Thread des Observables zum Thread des Observers. Der Entwickler muss sich darum nicht kümmern.
Wir haben gesehen, wie man Observables erstellt (Observable.from, Observable.create). Nun sehen wir uns die vordefinierten Observables der Bibliothek RxJava an.
7.3. Vordefinierte Observables
7.3.1. Beispiel-08: Die Methode [Observable.range]
 |  |
Von nun an werden wir spezielle Klassen für die beobachteten Prozesse und ihre Beobachter verwenden. Die Idee dahinter ist, deren Namen, Ausführungs-Thread und Ausführungszeiten zu protokollieren, um diese im Zeitverlauf nachverfolgen zu können.
Die Klasse [Process] ist einfach ein „Observable“, das man benennen kann. Sie implementiert die folgende Schnittstelle [IProcess]:
package dvp.rxjava.observables.utils;
import rx.Observable;
public interface IProcess<T> {
// Name des Beobachtungswerts
public String getName();
// Beobachtungsgröße
public Observable<T> getObservable();
}
Diese Schnittstelle kann von der folgenden Klasse [Process<T>] implementiert werden:
package dvp.rxjava.observables.utils;
import rx.Observable;
import rx.Scheduler;
public class Process<T> implements IProcess<T>{
// Name der Größe
protected String name;
// Beobachteter Prozess
protected Observable<T> observable;
// Konstruktoren
public Process(String name, Observable<T> observable) {
// lokale Initialisierungen
this.name = name;
this.observable = observable;
}
// Getter und Setter
public String getName() {
return name;
}
public Observable<T> getObservable() {
return observable;
}
}
- Zeile 9: Name des Prozesses;
- Zeile 11: die beobachtete Größe;
- Zeilen 14–18: der Konstruktor;
Der Beobachter wird durch die folgende Klasse [Observateur] beschrieben:
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> {
...
}
- Zeile 11: Die Klasse Observateur<T> erweitert die Klasse Subscriber<T>, die wir in Abschnitt 7.1.3 kurz vorgestellt haben. Wir werden sie als Argument der Methode [Observable.subscribe] verwenden:
// beobachtbare Ausführung (Beobachtung)
obs1.subscribe(observateur);
Die in Zeile 2 oben verwendete Methode [Observable.subscribe] hat folgende Definition:
 |
Die Aufgabe von [Subscriber] besteht hauptsächlich darin, die vom Observable ausgegebenen Elemente zu verwalten, das es abonniert hat, und zwar mithilfe der Methoden der Schnittstelle [Observer]: onNext, onError, onCompleted. Die Klasse [Subscriber] verfügt über folgende Methoden:
 |
Im Code der Klasse [Observateur] verwenden wir die Methode [1] isUnsubscribed, um festzustellen, ob das Abonnement des Abonnenten gekündigt wurde oder nicht. Die vollständige Klasse [Observateur<T>] lautet wie folgt:
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> {
// ein Sperrzeichen (Semaphor)
private CountDownLatch latch;
// eine Anzeigemethode
private Consumer<String> showInfos;
// Name des Beobachters
private String observerName;
// Name des beobachteten Prozesses
private String processName;
// Konstruktoren
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;
}
// --------------------------- Implementierung der Schnittstelle Observer<T>
@Override
public void onCompleted() {
// Ende der Übertragungen
if (!isUnsubscribed()) {
showInfos.accept(String.format("Subscriber [%s,%s].onCompleted", observerName, processName));
}
// Ende der Sperre des Haupt-Threads
latch.countDown();
}
@Override
public void onError(Throwable e) {
// Ausgabefehler
if (!isUnsubscribed()) {
showInfos.accept(String.format("Subscriber [%s, %s].onError (%s)", observerName, processName, e));
}
}
@Override
public void onNext(T value) {
// eine zusätzliche Emission
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));
}
}
}
}
- Zusätzlich zu den Merkmalen eines Subscriber wird der Beobachter Observateur die folgenden Informationen mit sich führen:
- Zeile 14: Ein Barrier oder Semaphor, der dazu dient, den Hauptthread zu blockieren, bis der Beobachter alle vom Beobachtbaren gesendeten Elemente empfangen hat. Dies geschieht in Zeile 36 des Codes, wenn der Beobachter vom Beobachtbaren die Benachrichtigung über das Ende der Übertragung erhält;
- Zeile 16: eine Instanz von Consumer<String>, die dazu dient, eine Meldung auf der Konsole anzuzeigen;
- Zeile 18: Der Name des Beobachters, um diese voneinander zu unterscheiden, wenn mehrere vorhanden sind;
- Zeile 20: der Name des beobachteten Prozesses;
- Zeilen 36, 46, 54: die Methoden [onCompleted, onError, onNext] der Schnittstelle [Observer<T>], die von der abstrakten Klasse [Subscriber<T>] implementiert wird. Diese Klasse implementiert sie nicht. Dies muss daher in den untergeordneten Klassen erfolgen. Bevor in diesen Methoden etwas unternommen wird, wird geprüft, ob der Beobachter nicht vom beobachteten Objekt abgemeldet wurde;
- Zeile 59: Die Methode [onNext] des Beobachters schreibt die Zeichenkette jSON des empfangenen Elements. Dies ermöglicht es uns, verschiedene Arten von Elementen anzuzeigen;
Vor diesem Hintergrund wollen wir nun eine neue Methode der Klasse Observable betrachten, nämlich die Methode [range]:
 |
Die Beobachtbare Observable.range(n,m) gibt (m) ganze Zahlen im Bereich von n bis n+m-1 aus. Wir untersuchen sie mit dem folgenden Code [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 {
// Anzahl der Beobachter
final int nbObservateurs = 2;
// Semaphor
CountDownLatch latch = new CountDownLatch(nbObservateurs);
// beobachtbare Konfiguration
Observable<Integer> obs1 = Observable.range(15, 3).subscribeOn(Schedulers.computation());
// beobachtbare Ausführung (Beobachtung)
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"));
}
// Warten
showInfos.accept("main : attente fin observation");
latch.await();
// Ende
showInfos.accept("main : fin observation");
}
// Anzeigen
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()));
}
- Zeile 16: Wir verwenden zwei Beobachter;
- Zeile 19: Der Barrierewächter (Semaphor) wird auf zwei initialisiert, da jeder Beobachter auf einem anderen Thread ausgeführt wird. Der Hauptthread muss daher das Ende der beiden Beobachtungs-Threads abwarten;
- Zeile 22: Wir konfigurieren das Observable so, dass es auf einem Thread des Schedulers [Schedulers.computation()] ausgeführt wird. Der Beobachter befindet sich auf demselben Thread wie das Observable;
- Zeilen 25–27: Wir abonnieren zwei Beobachter für das Observable. Dies löst für jeden der Beobachter die vollständige Ausführung des Observables aus: Die Ganzzahlen 15, 16 und 17 werden ausgegeben;
- Zeile 30: Der Hauptthread wartet auf das Ende der Beobachter;
Die erhaltenen Ergebnisse lauten wie folgt:
- Zeile 2: Der Hauptthread ist blockiert und wartet auf das Ende der beiden Beobachter;
- Zeilen 3–4: Man sieht, dass Beobachter 0 auf dem Thread [RxComputationThreadPool-1] und Beobachter 1 auf dem Thread [RxComputationThreadPool-2] läuft;
- Zeilen 3–10: Man sieht, dass beide Beobachter genau dieselben Elemente erhalten;
Wir werden die so definierte Klasse Observateur verwenden, um das Verhalten anderer Arten von Observables zu veranschaulichen.
7.3.2. Beispiel-09: Die Methoden von Observable.[interval, take, doNext]
 |
 |
Dieses Beispiel veranschaulicht die Verwendung des Observables Observable.interval (langes Intervall, Einheit TimeUnit), das in regelmäßigen Zeitabständen Long-Ganzzahlen ausgibt. Beachten Sie den Punkt [1]: Standardmäßig wird das Observable [Observable.interval] auf einem der Threads des Schedulers [Schedulers.computation] ausgeführt.
Der Code sieht wie folgt aus:
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 {
// Anzahl der Beobachter
final int nbObservateurs = 2;
// Semaphor
CountDownLatch latch = new CountDownLatch(nbObservateurs);
// beobachtbare Konfiguration
Observable<Long> obs1 = Observable.interval(500L, TimeUnit.MILLISECONDS).take(3)
.doOnNext(l -> showInfos.accept(l.toString()));
// beobachtbare Ausführung (Beobachtung)
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"));
}
// Warten
showInfos.accept("main : attente fin observation");
latch.await();
// Ende
showInfos.accept("main : fin observation");
}
// Anzeigen
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()));
}
- Zeile 22: Das Observable gibt alle 500 Millisekunden Long-Ganzzahlen aus. Die Reihe beginnt mit der Zahl 0;
- Zeile 22: Dieses Observable gibt eine unendliche Anzahl von Werten aus. Die Methode [Observable.take(n)] erstellt ein neues Observable, das nur die ersten n ausgegebenen Elemente beibehält;
 |
Kehren wir zum Code des Observables zurück:
Observable<Long> obs1 = Observable.interval(500L, TimeUnit.MILLISECONDS).take(3)
.doOnNext(l -> showInfos.accept(l.toString()));
Zeile 2: Die Methode [Observable.doOnNext] wird jedes Mal ausgeführt, wenn das Observable ein neues Element emittiert. Dies wird häufig zum Protokollieren von Informationen verwendet. Hier möchten wir das Emissionsdatum der Elemente protokollieren, um zu überprüfen, ob das Intervall von 500 Millisekunden eingehalten wird. Die Methode [Observable.doOnNext] verändert das Observable, auf das sie angewendet wird, nicht. Ihre Definition lautet wie folgt:
 |
Die Ausführung liefert folgende Ergebnisse:
- Zeilen 3, 7 und 11: Man sieht, dass das Sendeintervall annähernd 500 ms beträgt;
- Die beiden Beobachter laufen natürlich in zwei verschiedenen Threads, obwohl das Observable nicht für die Ausführung mit einem bestimmten Scheduler konfiguriert worden war. Hier sehen wir das Standardverhalten des Observables [Observable.interval];
7.3.3. Beispiele-10/12: Die Methoden von Observable.[error, empty, never]
 |  |
Wir werden die Methoden der Klasse [Observable] fortan prägnanter veranschaulichen. Der vorherige Code lautete wie folgt:
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 {
// Anzahl der Beobachter
final int nbObservateurs = 2;
// Semaphor
CountDownLatch latch = new CountDownLatch(nbObservateurs);
// beobachtbare Konfiguration
Observable<Long> obs1 = Observable.interval(500L, TimeUnit.MILLISECONDS).take(3)
.doOnNext(l -> showInfos.accept(l.toString()));
// beobachtbare Ausführung (Beobachtung)
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"));
}
// Warten
showInfos.accept("main : attente fin observation");
latch.await();
// Ende
showInfos.accept("main : fin observation");
}
// Anzeigen
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()));
}
Dieser Code wurde bereits im vorherigen Beispiel verwendet. Nur die Zeilen 21–22 änderten sich. Wir werden daher den Großteil dieses Codes in die folgende Klasse [ProcessUtils] auslagern:
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 {
// Semaphor
CountDownLatch latch = new CountDownLatch(nbObservateurs * processes.length);
// beobachtbare Ausführung (Beobachtung)
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()));
}
}
// Warten
showInfos.accept("main : attente fin observation");
latch.await();
// Ende
showInfos.accept("main : fin observation");
}
// Anzeigen
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()));
}
- Zeile 13: Die Methode akzeptiert zwei Parameter:
- nbObservateurs: die Anzahl der Beobachter der Prozesse, die als zweiter Parameter übergeben werden;
- processes: die zu beobachtenden Prozesse (benannte Observables). Dank der Notation [IProcess<?>] können die Prozesse Elemente unterschiedlicher Typen ausgeben;
- Zeile 16: Das Semaphor muss auf „grün“ wechseln, wenn alle Beobachter alle ihre Beobachtungen abgeschlossen haben. Der Anfangswert des Semaphors ist daher die Anzahl der Beobachter multipliziert mit der Anzahl der Beobachtungen;
- Zeilen 20–25: Jeder Beobachter wird auf alle zu beobachtenden Prozesse abonniert;
- Zeile 23: Die Beobachtungsgröße wird vom Prozess abgerufen (siehe Abschnitt 7.3.1);
- Zeile 23: Man meldet einen Beobachter dafür an. Man übermittelt diesem vier Informationen:
- seinen Namen;
- das Semaphor, das er dekrementieren muss, wenn er die Benachrichtigung über das Ende der Ausgabe der von ihm beobachteten Observable erhält;
- die Methode, die verwendet werden soll, wenn er Informationen auf der Konsole protokollieren möchte;
- den Namen des Prozesses, den er beobachten wird;
Nachdem diese Klassen definiert sind, sieht Beispiel 10 wie folgt aus:
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 {
// beobachtbare Konfiguration
Observable<?> obs = Observable.error(new RuntimeException("Erreur !!!")).subscribeOn(Schedulers.computation());
// beobachtbare Ausführung (Beobachtung)
ProcessUtils.subscribe(2,new Process<>("process1", obs));
}
}
In Zeile 11 wird die statische Methode [Observable.error] wie folgt definiert:
 |
Zeile 8 konfiguriert somit ein Observable, das lediglich eine Ausnahme an die Methode [onError] seiner Abonnenten auslöst. Die Ausführung liefert folgende Ergebnisse:
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]
In den Zeilen 3 und 4 hat die Methode [onError] der beiden Abonnenten die vom Observable ausgelöste Ausnahme empfangen.
Diese Ausführung weist eine Besonderheit auf: Die Methoden [onCompleted] der beiden Beobachter wurden nicht aufgerufen. Daher wurde die Barriere nicht gesenkt, und der Hauptthread bleibt in der statischen Methode [ProcessUtils.subscribe] in der folgenden Zeile 3 blockiert:
// Warten
showInfos.accept("main : attente fin observation");
latch.await();
// Ende
showInfos.accept("main : fin observation");
Hier stellt man fest, dass im Falle eines Fehlers des Observables die Methode [onCompleted] der Subscriber nicht aufgerufen wird. Wir ändern daher die Methode [Observateur.onError] wie folgt:
@Override
public void onError(Throwable e) {
// Sendefehler
if (!isUnsubscribed()) {
showInfos.accept(String.format("Subscriber[%s, %s].onError (%s)", observerName, processName, e));
}
// Ende der Blockierung des Haupt-Threads
latch.countDown();
}
Wir fügen die Zeilen 7–8 hinzu, um die Sperre im Falle eines Fehlers des Observables aufzuheben. Mit diesem neuen Code liefert die Ausführung folgende Ergebnisse:
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]
Wir erhalten Zeile 5, die wir zuvor nicht erhalten hatten.
Beispiel 11 lautet wie folgt:
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 {
// beobachtbare Konfiguration
Observable<?> obs1 = Observable.empty();
// Beobachtbare Ausführung (Beobachtung)
ProcessUtils.subscribe(2,new Process<>("process1",obs1));
}
}
In Zeile 10 erstellt die statische Methode [Observable.empty] ein Observable, das keine Elemente emittiert. Es emittiert lediglich die Benachrichtigung über das Ende der Emission;
 |
Die Ausführung des Codes aus dem obigen Beispiel liefert folgende Ergebnisse:
- Zeilen 2 und 3: Man sieht, dass beide Beobachter die Benachrichtigung über das Ende der Emission erhalten, ohne zuvor Elemente empfangen zu haben.
Man könnte sich fragen, wozu diese Methode eigentlich gut sein soll. Man kann sie analog zu einer Sammlung verwenden, die anfangs leer ist und in die anschließend Elemente hinzugefügt werden:
In Zeile 3 wird die anfängliche Beobachtungsgröße obs (Zeile 1) mit anderen Beobachtungsgrößen zusammengeführt.
Beispiel 12 veranschaulicht die statische Methode [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 {
// beobachtbare Konfiguration
Observable<?> obs1 = Observable.never();
// Ausführung (Beobachtung) beobachtbar
ProcessUtils.subscribe(2,new Process<>("process1",obs1));
}
}
Die statische Methode [Observable.never] erstellt eine Beobachtbare, die niemals emittiert:
 |
Die Ausführung des Beispiels liefert folgende Ergebnisse:
In Zeile 2 wartet der Hauptthread unbegrenzt. Tatsächlich sendet kein Observable die Benachrichtigung [onCompleted], die es ermöglicht, das Semaphor (die Schranke) auf Grün zu schalten (die Schranke zu senken).
7.4. Multi-threading
7.4.1. Beispiel 13: Aktions-Thread, Beobachtungs-Thread
In Abschnitt 7.1.3 haben wir mit der statischen Methode [Observable.create] ein Observable erstellt:
|
- Die Methode [create] gibt einen Typ Observable<T> zurück;
- Der Parameter der Methode [create] ist eine Funktion vom Typ [Observable.OnSubscribe<T>], die wie folgt definiert ist:
|
Der Typ [Observable.OnSubscribe<T>] ist eine funktionale Schnittstelle, die ihrerseits die funktionale Schnittstelle [Action1<Subscriber<? super T>>] erweitert. Die Methode [call] dieser Schnittstelle erwartet einen Typ [Subscriber] (Abonnent, Subscriber, Beobachter). Im weiteren Verlauf dieses Dokuments werden wir den Typ [Observable.OnSubscribe<T>] gelegentlich als Aktion bezeichnen. Wir werden benutzerdefinierte Aktionen erstellen, die einen Namen haben werden. Dabei handelt es sich um Instanzen der folgenden Schnittstelle [IProcessAction]:
 |
package dvp.rxjava.observables.utils;
import rx.Observable;
public interface IProcessAction<T> extends Observable.OnSubscribe<T> {
// Die Aktion hat einen Namen
public String getName();
}
- Zeile 5: Die Schnittstelle [IProcessAction<T>] weist alle Merkmale der Schnittstelle [Observable.OnSubscribe<T>] auf;
- Zeile 8: Sie verfügt zusätzlich über eine Methode [getName], die den Namen der Instanz zurückgibt, die die Schnittstelle implementiert;
Wir werden die folgende Aktion mit dem Namen [ProcessAction01] verwenden:
package dvp.rxjava.observables.utils;
import java.util.Random;
import rx.Subscriber;
import rx.functions.Func1;
public class ProcessAction01<T> implements IProcessAction<T> {
// Daten
private String name;
private int nbValues;
private Func1<Integer, T> func1;
// Konstruktoren
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++) {
// Wartezeit
try {
Thread.sleep(new Random().nextInt(500));
} catch (InterruptedException e) {
// Fehler
ProcessUtils.showInfos.accept(String.format("Observable (%s) onError", getName()));
subscriber.onError(e);
}
// Ausgabe eines Elements
T value = func1.call(i);
ProcessUtils.showInfos.accept(String.format("Observable (%s,%s) onNext (%s)", getName(), i, value));
subscriber.onNext(value);
}
// abgeschlossen
ProcessUtils.showInfos.accept(String.format("Observable (%s) onCompleted", getName()));
subscriber.onCompleted();
}
@Override
public String getName() {
return name;
}
}
- Zeile 8: Die Klasse [ProcessAction01<T>] implementiert die Schnittstelle [IProcessAction<T>] und somit auch die Schnittstelle [Observable.OnSubscribe<T>];
- Zeile 11: Name der Aktion;
- Zeile 12: die Anzahl der auszugebenden Werte;
- Zeile 13: Eine Instanz vom Typ [Func1<Integer, T>], die aus einer Ganzzahl einen Typ T erzeugt, der vom Observable ausgegeben wird (Zeilen 35 und 37);
- Zeilen 16–20: Dem Konstruktor werden der Name der Aktion, die Anzahl der auszugebenden Werte und die Ausgabefunktion übergeben;
- Zeilen 23–42: der Code des Prozesses;
- Zeile 23: Die Methode [call] erhält als Parameter den Abonnenten des mit dem Prozess verbundenen Observables;
- Zeile 28: Der Prozess sendet seine Elemente nach einer Wartezeit von zufälliger Dauer;
- Zeile 32: die Ausgabe eines Fehlers;
- Zeile 37: eine normale Emission;
- Zeile 41: Senden der Benachrichtigung über das Ende der Übertragung;
- Zeilen 25–38: Die Aktion sendet nach einer zufälligen Wartezeit (Zeile 30) reelle Werte an nbValues;
- Zeile 35: Der auszugebende Wert wird von der Funktion [func1] bereitgestellt, die als Parameter an den Konstruktor übergeben wird (Zeile 16);
Wir refaktorisieren die Klasse [Process] (siehe Abschnitt 7.3.1), damit sie auch mit einer benannten Aktion instanziiert werden kann. Wir fügen ihr den folgenden Konstruktor hinzu:
public Process(IProcessAction<T> na, Scheduler schedulerObserved, Scheduler schedulerObserver) {
// Prozessname=Aktionsname
name = na.getName();
// Aktion --> Beobachtungsgröße
observable = Observable.create(na);
// Ausführungsthread des beobachteten Prozesses
if (schedulerObserved != null) {
observable = observable.subscribeOn(schedulerObserved);
}
// Beobachtungs-Thread des Beobachters
if (schedulerObserver != null) {
observable = observable.observeOn(schedulerObserver);
}
}
- Zeile 1: Der Konstruktor akzeptiert 3 Parameter:
- die benannte Aktion, die zum Erstellen des Observables verwendet wird (Zeile 5);
- den Scheduler des beobachteten Prozesses (kann null sein);
- den Scheduler des Beobachters (kann z. B. null sein);
- Zeile 5: Die Beobachtbare wird anhand der als Parameter übergebenen Aktion erstellt;
Der folgende Code [Exemple13] beobachtet verschiedene Observables:
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 {
// Prozess 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 1, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// Prozess 2
Process<String> process2 = new Process<>(
new ProcessAction01<String>("process2", 2, i -> String.format("valeur-%s", i)), Schedulers.computation(), null);
// Prozess 3
Process<Integer> process3 = new Process<>(new ProcessAction01<Integer>("process3", 3, i -> i * 2), null,
Schedulers.computation());
// Prozess 4
Process<Boolean> process4 = new Process<>(new ProcessAction01<Boolean>("process4", 4, i -> i % 2 == 0), null, null);
// Abonnements
ProcessUtils.subscribe(1, process1);
ProcessUtils.subscribe(1, process2);
ProcessUtils.subscribe(1, process3);
ProcessUtils.subscribe(1, process4);
}
}
- Zeilen 13–15: Der Prozess process1 erzeugt eine reelle Zahl in einem Rechen-Thread, die in einem anderen Rechen-Thread beobachtet wird;
- Zeilen 17–18: Der Prozess process2 erzeugt zwei Zeichenketten in einem Rechen-Thread, wobei keine Angabe zum Thread des Beobachters gemacht wird. Die Ergebnisse zeigen, dass die Beobachtung standardmäßig auf demselben Thread erfolgt, auf dem der Prozess ausgeführt wird;
- Zeilen 20–21: Der Prozess process3 erzeugt 3 Ganzzahlen in einem nicht festgelegten Thread, die in einem Rechen-Thread beobachtet werden. Die Ergebnisse zeigen, dass die Ausführung des Prozesses standardmäßig im Haupt-Thread erfolgt;
- Zeile 23: Der Prozess process4 erzeugt 4 boolesche Werte auf einem nicht festgelegten Thread, die auf einem nicht festgelegten Thread beobachtet werden. Die Ergebnisse zeigen, dass sowohl die Ausführung des Prozesses als auch dessen Beobachtung standardmäßig auf dem Hauptthread erfolgen;
Das Ergebnis der Ausführung dieses Codes lautet wie folgt:
- Der Prozess process1 erzeugt 1 reelle Zahl (Zeile 4) auf dem Berechnungs-Thread [RxComputationThreadPool-4], die auf dem Berechnungs-Thread [RxComputationThreadPool-3] (Zeile 6) beobachtet wird;
- Der Prozess process2 erzeugt 2 Zeichenketten (Zeilen 12, 14) auf dem Berechnungs-Thread [RxComputationThreadPool-5], die auf demselben Thread beobachtet werden (Zeilen 13, 15);
- Der Prozess process3 erzeugt 3 Ganzzahlen (Zeilen 21, 23, 25) im Hauptthread, die im Berechnungs-Thread [RxComputationThreadPool-6] (Zeilen 22, 24, 28) beobachtet werden;
- Der Prozess process4 erzeugt im Hauptthread 4 boolesche Werte (Zeilen 34, 36, 38, 40), die im selben Hauptthread beobachtet werden (Zeilen 33, 35, 37, 39);
Der Leser wird gebeten, im obigen Text Folgendes zu verfolgen:
- den Lebenszyklus des beobachteten Prozesses und seines Threads;
- den Lebenszyklus seines Beobachters und dessen Thread;
Ein Großteil des Interesses an den Rx-Bibliotheken beruht auf diesem Multithreading, das der Entwickler nicht selbst verwalten muss.
7.5. Kombinationen mehrerer Observables
7.5.1. Beispiel 14: Zwei Observables mit [Observable.merge] zusammenführen
Wir stellen nun statische Methoden der Klasse [Observable] vor, mit denen sich mehrere Observables zu einem Ergebnis-Observable kombinieren lassen.
Das erste Beispiel dieser Art lautet wie folgt:
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 {
// Prozess 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// Prozess 2
Process<String> process2 = new Process<>(
new ProcessAction01<String>("process2", 2, i -> String.format("valeur-%s", i)), Schedulers.computation(), null);
// Zusammenführung
Process<?> process12 = new Process<>("process12",
Observable.merge(process1.getObservable(), process2.getObservable()));
// Abonnements
ProcessUtils.subscribe(1, process12);
}
}
- Zeilen 15–17: Ein Prozess namens [process1] gibt 3 reelle Zahlen auf einem Rechen-Thread aus. Er wird ebenfalls auf einem Rechen-Thread beobachtet;
- Zeilen 19–20: Ein Prozess namens [process2] gibt zwei Zeichenketten auf einem Rechen-Thread aus. Der Beobachtungs-Thread ist nicht vorgegeben. Wie bereits zuvor gesehen, ist in diesem Fall der Beobachtungs-Thread der Rechen-Thread;
- Zeile 23: Die beiden Prozesse werden zusammengeführt, d. h., es wird ein Observable erstellt, dessen Elemente gleichzeitig aus beiden Prozessen stammen. Dazu wird die statische Methode [Observable.merge] verwendet:
 |
Anders als das obige Schema vermuten lassen könnte, können sich bei der Zusammenführung die Elemente eines Stroms 1 zwischen die Elemente eines Stroms 2 schieben. Dies zeigen die Ergebnisse der Ausführung:
- Zeile 3: Der Prozess [process1] wird auf dem Rechen-Thread [RxComputationThreadPool-4] ausgeführt;
- Zeile 4: Der Prozess [process2] wird auf dem Rechen-Thread [RxComputationThreadPool-5] ausgeführt;
- Zeile 9: Der Prozess [process12] wird auf dem Rechen-Thread [RxComputationThreadPool-3] beobachtet. Ich kenne die Regel nicht, die zu dieser Auswahl geführt hat;
- Zeilen 9–11: Es ist zu sehen, dass der Beobachter Elemente der beiden Prozesse [process1] (Zeile 5) und [process2] (Zeilen 6, 7) beobachtet, obwohl keiner der beiden abgeschlossen ist (es liegt eine Vermischung vor);
- Der Prozess [process12] wird beendet (Zeile 17), sobald die beiden Prozesse process1 und process2 beendet sind;
7.5.2. Beispiel 15: Zwei Observables mit [Observable.concat] verknüpfen
Wir betrachten nun den folgenden Code:
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 {
// Prozess 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// Prozess 2
Process<String> process2 = new Process<>(
new ProcessAction01<String>("process2", 2, i -> String.format("valeur-%s", i)), null, Schedulers.computation());
// Verknüpfung
Process<?> process12 = new Process<>("process12",
Observable.concat(process1.getObservable(), process2.getObservable()));
// Abonnements
ProcessUtils.subscribe(1, process12);
}
}
- Zeilen 15–17: Ein Prozess namens [process1] gibt 3 reelle Zahlen an einen Berechnungs-Thread aus. Er wird ebenfalls auf einem Berechnungs-Thread beobachtet;
- Zeilen 19–20: Ein Prozess namens [process2] gibt zwei Zeichenketten an einen nicht festgelegten Thread aus, hier standardmäßig den Hauptthread. Er wird auf einem Rechen-Thread beobachtet;
- Zeile 23: Die beiden Prozesse werden verkettet, d. h., es wird ein Observable erstellt, dessen Elemente aus beiden Prozessen stammen. Es findet keine Vermischung der ausgegebenen Werte statt. Der Prozess [process12] gibt zunächst alle Werte des Prozesses [process1] und anschließend die des Prozesses [process2] aus. Dazu wird die statische Methode [Observable.concat] verwendet:
 |
Die Ergebnisse der Ausführung lauten wie folgt:
- Zeilen 3–10: Der Prozess [process1] wird ausgeführt, und der Prozess [process12] gibt die von [process1] ausgegebenen Werte weiter;
- Zeile 9: Der Prozess [process1] ist beendet;
- Zeilen 11–17: Der Prozess [process2] wird ausgeführt, und der Prozess [process12] gibt die von [process2] ausgegebenen Werte weiter;
Beim Prozess process2 gibt es eine Besonderheit: Es wurde kein Ausführungsthread vorgegeben. Man hätte daher erwarten können, dass standardmäßig der Hauptthread verwendet wird. Dies ist jedoch nicht der Fall. Der Ausführungsthread war der Berechnungsthread [RxComputationThreadPool-3] (Zeile 11). Wenn also weder ein Ausführungsthread noch ein Beobachtungsthread vorgegeben wird, kann man keine Annahme darüber treffen, welcher Thread ausgewählt wird.
7.5.3. Beispiel 16: Kombinieren von zwei Observablen mit [Observable.zip]
Wir betrachten nun den folgenden Code:
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 {
// Prozess 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// Prozess 2
Process<String> process2 = new Process<>(
new ProcessAction01<String>("process2", 2, i -> String.format("valeur-%s", i)), null, null);
// Funktion zur Kombination der beiden Prozesse
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");
}
}
};
// ZIP-Datei der beiden Prozesse
Process<String> process12 = new Process<>("process12",
Observable.zip(Arrays.asList(process1.getObservable(), process2.getObservable()), funcn));
// Abonnements
ProcessUtils.subscribe(1, process12);
}
}
- Zeilen 16–18: Ein Prozess namens [process1] gibt 3 reelle Zahlen auf einem Berechnungs-Thread aus. Er wird ebenfalls auf einem Berechnungs-Thread beobachtet;
- Zeilen 20–21: Ein Prozess namens [process2] gibt zwei Zeichenketten an einen nicht festgelegten Thread aus. Der Beobachtungs-Thread ist ebenfalls nicht festgelegt;
- Zeilen 23–32: Instanziierung eines Typs [FuncN<String>] mit einer anonymen Klasse. FuncN ist eine funktionale Schnittstelle:
 |
Die Methode [FuncN.call] erwartet ein Array von Objekten und gibt einen Typ R zurück. Die Funktion [funcn] wird verwendet, um die Prozesse process1 und process2 in dieser Reihenfolge zu kombinieren. In der Methode [FuncN.call]:
- ist `args[0]` ein `Double`;
- args[1] ist ein String;
Hier ist das Ergebnis von [funcn.call] die Zeichenkette aus Zeile 27. Für die Ermittlung dieses Ergebnisses ist es nicht erforderlich, die Typen der Argumente der Methode call zu kennen.
Die beiden Prozesse werden wie folgt kombiniert:
// ZIP-Datei der beiden Prozesse
Process<String> process12 = new Process<>("process12",
Observable.zip(Arrays.asList(process1.getObservable(), process2.getObservable()), funcn));
Die Methode [Observable.zip] funktioniert wie folgt:
 |
Es ist ersichtlich, dass:
- das erste Argument von „zip“ ein Iterable<Observable> ist. In unserem Beispiel haben wir einen effektiven Parameter vom Typ List<Observable>, der sich aus unseren beiden Observablen zusammensetzt;
- das zweite Argument von `zip` ist vom Typ `FuncN`. In unserem Beispiel lautet der effektive Parameter `[funcn]`;
Die Ausführung liefert folgende Ergebnisse:
- Zeilen 7, 11: Der Prozess process12 gibt zwei Elemente aus;
- Zeile 8: Das zusätzliche Element, das vom Prozess process1 ausgegeben wird und im Prozess process2 keinen Partner hat, wird vom Ergebnisprozess process12 nicht ausgegeben;
Es ist ersichtlich, dass der Prozess process2, dem weder ein Ausführungs- noch ein Beobachtungs-Thread zugewiesen worden war, den Haupt-Thread für beide verwendet hat.
7.5.4. Beispiel 17: Kombination zweier Beobachtungsgrößen mit [Observable.combineLatest]
Wir betrachten nun den folgenden Code:
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 {
// Prozess 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// Prozess 2
Process<Double> process2 = new Process<>(
new ProcessAction01<Double>("process2", 2, i -> new Random().nextInt(200) * 1.4), null,
Schedulers.computation());
// Kombination der beiden Prozesse
Process<Double> process12 = new Process<>("process12",
Observable.combineLatest(process1.getObservable(), process2.getObservable(), (d1, d2) -> d1 + d2));
// Abonnements
ProcessUtils.subscribe(1, process12);
}
}
- Zeilen 14–16: Ein Prozess namens [process1] gibt 3 reelle Zahlen auf einem Berechnungs-Thread aus. Er wird ebenfalls auf einem Berechnungs-Thread beobachtet;
- Zeilen 18–20: Ein Prozess namens [process2] gibt zwei reelle Zahlen an einen nicht festgelegten Thread aus. Diese werden auf einem Rechen-Thread beobachtet;
- Zeile 23: Die beiden Observablen werden mit der folgenden statischen Methode „[Observable.combineLatest]“ kombiniert:
 |
Die Beobachtungsgröße [combineLatest] funktioniert wie folgt: Wenn eine der beiden Beobachtungsgrößen ein Element E1 ausgibt, wird dieses Element durch [combineFunction] mit dem zuletzt von der anderen Beobachtungsgröße ausgegebenen Element kombiniert.
Die Ausführung dieses Codes liefert das folgende Ergebnis:
- Zeile 5: Die Ausgabe von process2 (56) wird mit dem zuletzt von process1 (54, Zeile 4) ausgegebenen Element kombiniert und liefert das Ergebnis in Zeile 7;
- Zeile 6: Die Ausgabe von process1 (51,6) wird mit dem zuletzt von process2 (56, Zeile 5) ausgegebenen Element kombiniert und ergibt das Ergebnis in Zeile 8;
- Zeile 9: Die Ausgabe von process2 (261,8) wird mit dem zuletzt von process1 (51,6, Zeile 6) ausgegebenen Element kombiniert und ergibt das Ergebnis von Zeile 12;
- Zeile 13: Die Ausgabe von process1 (80,39) wird mit dem zuletzt von process2 (261,8, Zeile 9) ausgegebenen Element kombiniert und ergibt das Ergebnis von Zeile 15;
Hier handelt es sich um eine Variante des Observables [zip], bei der die kombinierten Elemente diesmal nicht unbedingt die Elemente an derselben Position in den Datenströmen sind. Es ist hier zu beachten, dass der Prozess process2, für den kein Ausführungs-Thread vorgegeben wurde, hier auf dem Haupt-Thread ausgeführt wurde (Zeile 2).
7.5.5. Beispiel 18: Kombinieren von zwei Observables mit [Observable.amb]
Wir betrachten nun den folgenden Code:
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 {
// Prozess 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// Prozess 2
Process<Double> process2 = new Process<>(
new ProcessAction01<Double>("process2", 2, i -> new Random().nextInt(200) * 1.4), null, null);
// Kombination der beiden Prozesse
Process<Double> process12 = new Process<>("process12",
Observable.amb(process1.getObservable(), process2.getObservable()));
// Abonnements
ProcessUtils.subscribe(1, process12);
}
}
- Zeilen 14–16: Ein Prozess namens [process1] gibt 3 reelle Zahlen auf einem Berechnungs-Thread aus. Er wird ebenfalls auf einem Berechnungs-Thread beobachtet;
- Zeilen 18–20: Ein Prozess namens [process2] gibt zwei reelle Zahlen an einen nicht festgelegten Thread aus. Diese werden auf einem nicht festgelegten Thread beobachtet;
- Zeile 22: Die beiden Observablen werden mit der folgenden statischen Methode [Observable.amb] kombiniert:
 |
Wie das obige Schema zeigt, gibt die Beobachtbare „[Observable.amb(Observable o1, Observable o2)]“ die Elemente der Beobachtbaren aus, die als erste Werte ausgibt. Dies wird durch die Ergebnisse des vorgestellten Beispiels bestätigt:
- Zeile 4: Der Prozess process2 sendet als Erster;
- Zeilen 8, 12: Der Prozess process12 gibt alle Elemente aus, die vom Prozess process2 ausgegeben wurden (Zeilen 4, 11);
7.6. Verarbeitungskette eines Observables
7.6.1. Beispiel 19: Umwandlung einer Beobachtbaren mit [Observable.map]
In den vorangegangenen Beispielen haben wir verschiedene Kombinationen von zwei Observables zu einem dritten Observable untersucht. Nun stellen wir statische Methoden der Klasse [Observable] vor, die Transformations-, Filter- und Aggregationsoperationen auf ein Observable ermöglichen. Hier finden wir Methoden, die denen der in Abschnitt 5 behandelten Klasse [Stream] entsprechen.
Unser erstes Beispiel lautet wie folgt:
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 {
// Prozess 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// Prozess 2
Process<String> process2 = new Process<>("process2",
process1.getObservable().map(d -> String.format("valeur-%s", d)));
// Abonnements
ProcessUtils.subscribe(1, process2);
}
}
- Zeilen 14–16: Ein Prozess namens process1 gibt 3 reelle Zahlen auf einem Berechnungsthread aus. Er wird ebenfalls auf einem Berechnungsthread beobachtet;
- Zeilen 17–18: Die von process1 ausgegebenen Zahlen werden in einem Prozess namens process2 in Zeichenketten umgewandelt;
- Zeile 20: process2 wird beobachtet;
Die Methode [Observable.map] in Zeile 18 entspricht der in Abschnitt 5.5 behandelten Methode [Stream.map]:
 |
Die Ergebnisse des Beispiels lauten wie folgt:
- Zeilen 4, 5 und 8: die Ausgabewerte von process1. Es handelt sich um reelle Zahlen;
- Zeilen 6, 7, 10: die Emissionen von process2, die beobachtet wurden. Dies sind Zeichenfolgen;
7.6.2. Beispiel-20: Filtern einer Beobachtungsgröße mit [Observable.filter]
Das Beispiel sieht wie folgt aus:
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 {
// Prozess 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// Prozess 2
Process<Integer> process2 = new Process<>("process2", process1.getObservable().filter(i -> i % 2 == 0));
// Abonnements
ProcessUtils.subscribe(1, process2);
}
}
- Zeilen 11–12: Ein Prozess namens process1 gibt die ganzen Zahlen von 0 bis 2 auf einem Rechen-Thread aus. Er wird ebenfalls auf einem Rechen-Thread beobachtet;
- Zeile 14: Die von process1 ausgegebenen Zahlen werden gefiltert, sodass in process2 nur die geraden Zahlen verbleiben;
- Zeile 20: process2 wird beobachtet;
Die Methode [Observable.filter] in Zeile 18 entspricht der in Abschnitt 5.4 behandelten Methode [Stream.filter]:
 |
Die Ergebnisse des Beispiels lauten wie folgt:
- Zeilen 4, 5 und 7: die Sendungen von process1;
- Zeilen 6, 9: die beobachteten Emissionen von process2. Es handelt sich um die geraden Elemente von process1;
7.6.3. Beispiel 21: Umwandlung einer Beobachtung mit [Observable.flatMap]
Das Beispiel lautet wie folgt:
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 {
// Prozess 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// Prozess 2
Process<Integer> process2 = new Process<>("process2", process1.getObservable().flatMap(i -> {
int value = i * 10;
return Observable.just(value, value + 1, value + 2);
}));
// Abonnements
ProcessUtils.subscribe(1, process2);
}
}
- Zeilen 12–13: Ein Prozess namens process1 gibt die ganzen Zahlen von 0 bis 2 auf einem Rechen-Thread aus. Er wird ebenfalls auf einem Rechen-Thread beobachtet;
- Zeilen 15–18: Jede von process1 ausgegebene Zahl n wird in eine Beobachtbare umgewandelt, die die drei Zahlen (10*n, 10*n+1, 10*n+2) ausgibt. Würde in Zeile 15 die Methode [map] verwendet, würde process2 einen Typ Observable<Integer> und nicht einen Typ Integer emittieren. Die verwendete Methode [flatMap] ermöglicht es, (flatten) diese Folge von Elementen vom Typ Observable<Integer> in eine Folge von Elementen vom Typ Integer umzuwandeln, die aus jedem der Elemente jedes einzelnen Observable<Integer> besteht;
- Zeile 20: Es wird process2 beobachtet;
Die Methode [Observable.flatMap] in Zeile 15 entspricht der in Abschnitt 5.6.12 behandelten Methode [Stream.flatMap]:
 |
Die Ergebnisse des Beispiels lauten wie folgt:
- Zeilen 5–7: die drei Übertragungen von process2 im Anschluss an die Übertragung der Zeile 4 von process1;
- Zeilen 9–11: die drei Ausgabewerte von process2 im Anschluss an die Ausgabe von Zeile 8 von process1;
- Zeilen 14–16: die drei Sendungen von process2 im Anschluss an die Sendung in Zeile 12 von process1;
Der folgende Code zeigt, wie man einen Typ Observable<Integer[]> aus process1 und [Exemple21b] erstellt:
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 {
// Prozess 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// Prozess 2
Process<Integer[]> process2 = new Process<>("process2", process1.getObservable().map(i -> {
int value = i * 10;
return new Integer[] { value, value + 1, value + 2 };
}));
// Abonnements
ProcessUtils.subscribe(1, process2);
}
}
- Zeile 14: Es wird die Methode [Observable.map] verwendet;
- Zeile 16: Diese gibt einen Typ Integer[] zurück;
Die Ergebnisse lauten wie folgt:
- Zeilen 6, 7, 10: Hier sind die Ergebnisse von map zu sehen;
Alle diese Transformationen von Observables können verkettet werden, da jede Transformation ein neues Observable erzeugt. Dies zeigt das folgende Beispiel [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 {
// Prozess 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// Prozess 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));
// Abonnements
ProcessUtils.subscribe(1, process2);
}
}
- Zeilen 15–18: Auf das flatMap folgt ein filter;
Die Ausführungsergebnisse lauten wie folgt:
- Zeilen 8–13: process2 hat nur die geradzahligen Elemente aus flatMap ausgegeben;
Eine dem [flatMap] ähnliche Methode ist die Methode [flatMapIterable], die durch das folgende Beispiel [Exemple21d] veranschaulicht wird:
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 {
// Prozess 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// Prozess 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));
// Abonnements
ProcessUtils.subscribe(1, process2);
}
}
In Zeile 16 wird anstelle der Methode [flatMap] die Methode [flatMapIterable] verwendet. In diesem Fall muss die Transformationsfunktion einen Typ Iterable<T> (Zeile 18) anstelle eines Typs Observable<T> erzeugen.
Man erhält dieselben Ergebnisse wie zuvor.
Kehren wir zur Definition der Methode [flatMap] zurück:
 |
Wie oben zu sehen ist, hat sich ein blaues Element [3] zwischen die beiden grünen Elemente [1-2] eingefügt. Das bedeutet, dass die Methode [flatMap] bei der Reduzierung der Observable<T> die Ausgabereihenfolge dieser verschiedenen internen Observablen beibehält. Dies wird durch das folgende Beispiel [Exemple21e] veranschaulicht:
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 {
// Prozess 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 2, i -> i), Schedulers.computation(),
Schedulers.computation());
// Prozess 2
Process<Integer> process2 = new Process<>(new ProcessAction01<>("process2", 3, i -> i + 10),
Schedulers.computation(), Schedulers.computation());
// Prozess 3
Process<Integer> process3 = new Process<>("process3",
process1.getObservable().flatMap(i -> process2.getObservable()));
// Abonnements
ProcessUtils.subscribe(1, process3);
}
}
- Zeilen 11–12: Der Prozess process1 gibt die Ganzzahlen [0,1] aus;
- Zeilen 14–15: Der Prozess process2 gibt die ganzen Zahlen [10,11,12] aus;
- Zeilen 17–18: Jedem von process1 ausgegebenen Element wird die Beobachtungsgröße des Prozesses process2 zugeordnet. Das bedeutet:
- dem Element [0] von Prozess1 wird eine Beobachtungsgröße zugeordnet, die die [10,11,12] ausgibt;
- das Gleiche gilt für das Element 1;
Letztendlich werden die 6 Zahlen [10, 11, 12, 10, 11, 12] ausgegeben. Wir wollen sehen, in welcher Reihenfolge.
Die Ergebnisse der Ausführung lauten wie folgt:
Man sieht, dass die Ausgabereihenfolge des Prozesses process3 wie folgt war: [10, 10, 11, 12, 11, 12] (Zeilen 11, 12, 14, 17, 19, 22). Es ist also tatsächlich zu einer Vermischung der vom Prozess process2 ausgegebenen Elemente gekommen. Dies lässt sich vermeiden, indem man anstelle der Methode [flatMap] die Methode [concatMap] verwendet. Dies zeigt der folgende Code [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 {
// Prozess 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 2, i -> i), Schedulers.computation(),
Schedulers.computation());
// Prozess 2
Process<Integer> process2 = new Process<>(new ProcessAction01<>("process2", 3, i -> i + 10),
Schedulers.computation(), Schedulers.computation());
// Prozess 3
Process<Integer> process3 = new Process<>("process3",
process1.getObservable().concatMap(i -> process2.getObservable()));
// Abonnements
ProcessUtils.subscribe(1, process3);
}
}
In Zeile 18 wurde [flatMap] durch [concatMap] ersetzt. Die Ergebnisse der Ausführung lauten wie folgt:
Es ist ersichtlich, dass die Ausgabereihenfolge des Prozesses process3 wie folgt war: [10, 11, 12, 10, 11, 12] (Zeilen 12–14, 17, 19, 22). Die vom Prozess process2 ausgegebenen Elemente wurden nicht gemischt.
Eine weitere Variante der Methode [map] ist die Methode [switchMap]:
 |
Aus dem Observable [1] entstehen oben drei weitere Observables [2] mit jeweils zwei Elementen, die anschließend wie bei [flatMap] und [3] abgeflacht werden. Man kann feststellen, dass das Ergebnis 5 Elemente und nicht 6 hat. Dies liegt daran, dass, bevor die zweite Beobachtungsgröße ihr Element Nr. 2 [6] ausgibt, die dritte Beobachtungsgröße bereits ihr erstes Element [5] ausgibt, wodurch die zweite Beobachtungsgröße verworfen wird. Das Element [6] ist daher im resultierenden Observablen [3] nicht enthalten.
Zur Veranschaulichung von [switchMap] verwenden wir das folgende Beispiel [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 {
// Prozess 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 2, i -> i), Schedulers.computation(),
Schedulers.computation());
// Prozess 2
Process<Integer> process2 = new Process<>(new ProcessAction01<>("process2", 3, i -> i + 10),
Schedulers.computation(), Schedulers.computation());
// Prozess 3
Process<Integer> process3 = new Process<>("process3",
process1.getObservable().switchMap(i -> process2.getObservable()));
// Abonnements
ProcessUtils.subscribe(1, process3);
}
}
Die Ausführung des Beispiels liefert folgende Ergebnisse:
- process1 sendet 2 Elemente aus, aus denen 2 Beobachtbare process2 mit jeweils 3 Elementen entstehen;
- Zeile 14: Der Beobachter empfängt das Element Nr. 0, das vom ersten Beobachtbaren process2 in Zeile 6 gesendet wurde;
- Zeile 15: Der Beobachter empfängt das Element Nr. 0, das von der zweiten Beobachtungsgröße process2 in Zeile 13 gesendet wurde. Es ist nicht bekannt, warum er zuvor die Elemente 1 und 2 nicht empfangen hat, die vom ersten Beobachtbaren process2 in den Zeilen 7 und 8 gesendet wurden. Jedenfalls wird das erste Beobachtbare process2 verworfen;
- letztendlich sieht der Beobachter nur 4 Elemente (Zeilen 14, 15, 17, 20) statt der 6, die gesendet wurden;
7.6.4. Beispiele-22: Weitere Methoden der Klasse [Observable]
Die Klasse [Observable] übernimmt zahlreiche Methoden der Klasse [Stream] mit ähnlicher Funktionsweise. Hier sind einige davon. Wir beschränken uns darauf, den Code und die Ergebnisse anzugeben.
[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 {
// Prozess
Process<Integer> process = new Process<>("process", Observable.range(1, 10).take(3));
// Abonnements
ProcessUtils.subscribe(1, process);
}
}
Ergebnisse
[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 {
// Prozess
Process<Integer> process = new Process<>("process", Observable.range(1, 10).takeLast(2));
// Abonnements
ProcessUtils.subscribe(1, process);
}
}
Ergebnisse
[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 {
// Prozesse
Process<Integer> process = new Process<>("process", Observable.range(1, 10).skip(5).take(2));
// Abonnements
ProcessUtils.subscribe(1, process);
}
}
Ergebnisse
[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 {
// Prozesse
Process<Integer> process = new Process<>("process", Observable.range(1, 10).reduce(0, (i, a) -> i + a));
// Abonnements
ProcessUtils.subscribe(1, process);
}
}
- Zeile 10: Berechnet die Summe der Elemente der Observable. Das Ergebnis ist eine Observable, die diese Summe ausgibt;
Ergebnisse
[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 {
// Prozesse
Process<Boolean> process = new Process<>("process", Observable.range(1, 10).all(i -> i > 10));
// Abonnements
ProcessUtils.subscribe(1, process);
}
}
- Zeile 10: Gibt ein Observable<Boolean> zurück, das das Element true ausgibt, wenn das Prädikat der Methode [all] für alle Elemente wahr ist, andernfalls false;
Ergebnisse
[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 {
// Prozesse
Process<Integer> process = new Process<>("process", Observable.range(1, 10).count());
// Abonnements
ProcessUtils.subscribe(1, process);
}
}
- Zeile 10: [Observable.count] erstellt ein Observable mit einem Element, das der Summe der beobachteten Elemente entspricht;
Ergebnisse
[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 {
// Prozesse
Process<Integer> process = new Process<>("process", Observable.just(1, 2, 1, 3).distinct());
// Abonnements
ProcessUtils.subscribe(1, process);
}
}
Ergebnisse
[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 {
// Prozesse
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()));
// Abonnements
ProcessUtils.subscribe(1, process);
}
}
- Zeile 11: Die Methode [groupBy] gruppiert die 10 ausgegebenen Elemente in zwei Gruppen: die geraden und die ungeraden Zahlen. Das Ergebnis ist ein Typ „Observable<GroupedObservable<Boolean, Integer>>“, d. h. ein Observable, dessen Elemente vom Typ „GroupedObservable<Boolean, Integer>“ sind, wobei „Boolean“ der Typ des Gruppenschlüssels ist (hier false, true) ist und der auch der Typ des Ergebnisses des Lambda-Ausdrucks ist, der als Parameter an die Methode [groupBy] übergeben wird, und Integer der Typ der Elemente der Gruppe ist;
- Zeile 12: Der Typ GroupedObservable verfügt über eine Methode [asObservable], mit der sich aus diesem Typ ein Observable erstellen lässt. Wir erhalten also zwei Typen Observable<Integer>, einen für gerade Zahlen und einen für ungerade Zahlen. Aus diesen beiden Observables erstellt die Methode [concatMap] ein einziges;
Ergebnisse
[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 {
// Prozess 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// Prozess 2
Process<Timestamped<Integer>> process2 = new Process<>("process2", process1.getObservable().timestamp());
// Abonnements
ProcessUtils.subscribe(1, process2);
}
}
- In Zeile 15 ordnet die Methode [timestamp] jedem Element der verarbeiteten Beobachtungsgröße eine Uhrzeit zu;
Ergebnisse
In diesem Beispiel ist es schwer zu sagen, was die Information timestamp darstellt:
- Zeilen 4–5: Man sieht, dass Element 1 von process1 139 ms nach Element 0 gesendet wurde;
- Zeilen 6 und 7: Man sieht, dass Element 1 von process2 234 ms nach Element 0 beobachtet wurde;
- Zeilen 5, 8: Man sieht, dass Element 2 von process1 33 ms nach Element 1 gesendet wurde;
- Zeilen 7 und 10: Hier ist zu sehen, dass Element 2 von process2 37 ms nach Element 1 beobachtet wurde;
Diese Verzögerungen sind darauf zurückzuführen, dass die Threads für die Beobachtung und die Ausführung der Observablen nicht identisch sind. Ersetzt man die Zeilen 12–13 durch die folgenden Zeilen (Beispiel 22j):
// Prozess 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
null);
- Zeilen 2–3: Der Beobachtungs-Thread wird nicht vorgegeben. Man weiß, dass in diesem Fall die Beobachtungsgröße dort beobachtet wird, wo sie ausgeführt wird;
Dies führt zu folgenden Ergebnissen:
- Zeilen 4 und 6: Der Prozess process1 sendet sein Element Nr. 1 587 ms nach seinem Element Nr. 0;
- Zeilen 5 und 7: Der Beobachter beobachtet diese beiden Elemente im Abstand von 586 ms;
- Zeilen 6 und 8: Der Prozess process1 sendet sein Element Nr. 2 396 ms nach seinem Element Nr. 1;
- Zeilen 7 und 9: Der Beobachter beobachtet diese beiden Elemente im Abstand von 396 ms;
Hier sind die Werte von timestamp konsistent: Sie geben tatsächlich das Sendedatum des Elements wieder.
7.7. Die Scheduler
7.7.1. Beispiel 23: Der Scheduler [Schedulers.computation]
Wir betrachten nun die Ausführungsplaner. Die Beobachtung erfolgt im Ausführungsthread.
Das Thema der Scheduler ist etwas undurchsichtig. Die verschiedenen Scheduler werden in dieser Frage auf der Website von StackOverflow [http://stackoverflow.com/questions/31276164/rxjava-schedulers-use-cases] vorgestellt:
 |
Wir werden versuchen, die Verwendung dieser verschiedenen Scheduler anhand von Beispielen zu veranschaulichen. Das erste Beispiel veranschaulicht den Scheduler [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 {
// Prozesse
@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);
}
// Abonnements
ProcessUtils.subscribe(1, processes);
}
}
- Zeilen 14–19: Es wird ein Array mit 10 Prozessen erstellt, die auf einem Rechen-Thread ausgeführt werden;
- Zeile 17: Jeder Prozess gibt eine zufällige reelle Zahl aus;
- Zeile 21: Man abonniert alle diese Prozesse;
Die Ergebnisse lauten wie folgt:
- Zeilen 2–10: Die ersten 8 Prozesse starten auf 8 verschiedenen Threads (der verwendete Rechner hat 8 Kerne). Man kann feststellen, dass sie alle ungefähr zur gleichen Zeit starten;
- Zeilen 17–19: 3 Prozesse werden beendet und geben somit 3 Threads frei;
- Zeilen 23–24: Die beiden letzten Prozesse können nun starten, indem sie zwei der freigewordenen Threads nutzen;
Zusammenfassend lässt sich also festhalten, dass der Scheduler [Schedulers.computation] einen Pool von n Threads bereitstellt, wobei n die Anzahl der Kerne des Rechners ist. Die Threads werden parallel auf diesen Kernen ausgeführt.
7.7.2. Beispiel 24: Der Scheduler [Schedulers.io]
Wir lassen den vorherigen Code mit dem Scheduler [Schedulers.io] ausführen:
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 {
// Prozesse
@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);
}
// Abonnements
ProcessUtils.subscribe(1, processes);
}
}
- Zeile 18: Die Prozesse werden mit den Threads des Schedulers [Schedulers.io] ausgeführt;
Dies führt zu folgenden Ergebnissen:
- Zeilen 2–10: Die 10 Prozesse werden jeweils in einem eigenen Thread gestartet. Im Gegensatz zum vorherigen Fall konnten alle Prozesse gestartet werden. Es fällt auf, dass diese Startvorgänge 6 ms dauern, während es zuvor 1 ms waren;
- Zeilen 13–18: Die Observables senden nacheinander und nicht quasi parallel, wie es zuvor der Fall war;
Was ist der Unterschied zwischen den Schedulern [Schedulers.io] und [Schedulers.computation]? Eine Antwort findet sich unter URL [http://stackoverflow.com/questions/31276164/rxjava-schedulers-use-cases]:
 |
7.7.3. Beispiel 25: Der Scheduler [Schedulers.newThread]
Wir lassen den vorherigen Code mit dem Scheduler [Schedulers.newThread] ausführen:
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 {
// Prozesse
@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);
}
// Abonnements
ProcessUtils.subscribe(1, processes);
}
}
Die Ergebnisse sind dieselben wie beim Scheduler [Schedulers.io]:
In den Artikeln zu URL und [http://stackoverflow.com/questions/33415881/retrofit-with-rxjava-schedulers-newthread-vs-schedulers-io] wird erläutert, dass der Scheduler [Schedulers.io] einen Thread-Pool bereitstellt, was beim Scheduler [Schedulers.newThread] nicht der Fall ist. Ein Thread-Pool erstellt automatisch eine Anzahl n von Threads. Er weist diese den Prozessen zu, die sie benötigen. Wenn diese abgeschlossen sind, werden ihre Threads nicht gelöscht, sondern kehren in den Pool zurück und können dann von einem anderen Prozess wiederverwendet werden. Dies ist effizienter, als ständig Threads zu erstellen und zu löschen. Man kann daher davon ausgehen, dass es vorteilhafter ist, den Scheduler [Schedulers.io] zu verwenden.
7.7.4. Beispiel 26: Die Scheduler [Schedulers.immediate, Schedulers.trampoline]
Kehren wir zur Erläuterung dieser beiden Scheduler zurück:
|
Die Erklärung ist recht einfach zu verstehen, aber wenn man sie veranschaulichen will, merkt man, dass man sie nicht verstanden hat. Dank des Buches [Learning Reactive Programming With Java 8] konnte ich ein Beispiel erstellen, das auf einem in diesem Buch gefundenen Beispiel basiert, dieses jedoch vereinfacht. Es lautet wie folgt:
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 {
// ein Scheduler
Scheduler scheduler = Schedulers.immediate();
// ein Worker dieses Schedulers
Worker worker = scheduler.createWorker();
// ein Typ „Action0“, der auf dem Worker ausgeführt werden soll
Action0 action02 = new Action0() {
@Override
public void call() {
// Protokoll „action02“
ProcessUtils.showInfos.accept("action02");
}
};
// ein Typ „Action0“, der auf dem Worker ausgeführt werden soll
Action0 action01 = new Action0() {
@Override
public void call() {
// Es wird eine neue Aktion auf demselben Worker programmiert
worker.schedule(action02);
// Protokoll der Aktion 01
ProcessUtils.showInfos.accept("action01");
}
};
// action01 ist auf dem Worker geplant
worker.schedule(action01);
}
// Anzeigen
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()));
}
- Zeile 17: ein Scheduler. Das wird entweder [Schedulers.immediate] wie hier oder später [Schedulers.trampoline] sein;
- Zeile 19: Man kann Aktionen vom Typ Action0 (Zeilen 21, 20) auf den Workern des Schedulers ausführen lassen. Mit der Methode [Scheduler.createWorker] lässt sich ein Worker erstellen. Mit der Methode [Worker.schedule(Action0)] lässt sich ein Typ Action0 von einem Worker ausführen;
- Zeilen 21–27: Eine erste Aktion namens [action02], die (in Zeile 40) vom Worker aus Zeile 19 ausgeführt wird;
- Zeilen 30–38: Eine zweite Aktion namens [action01]. Sie zeichnet sich dadurch aus, dass sie die Aktion action02 auf demselben Worker ausführt wie sie selbst (Zeile 34). Hier liegt der Unterschied zwischen [Schedulers.immediate] und [Schedulers.trampoline]:
- Wenn der Scheduler [Schedulers.immediate] ist, wird in Zeile 34 die Aktion action02 sofort ausgeführt (daher der Name des Schedulers) und die gerade laufende Aktion action01 wird unterbrochen. Daraufhin wird die Meldung aus Zeile 25 angezeigt. Sobald die Aktion action02 beendet ist, wird die Aktion action01 fortgesetzt und die Meldung aus Zeile 36 angezeigt;
- Wenn der Scheduler [Schedulers.trampoline] ist, wird in Zeile 34 die Aktion action02 in die Warteschlange gestellt. Sie wird erst ausgeführt, wenn die laufende Aufgabe action01 beendet ist. Dann wird die Meldung in Zeile 36 angezeigt. Sobald die Aktion action01 abgeschlossen ist, wird die Aktion action02 ausgeführt, und wir sehen die Meldung in Zeile 25;
Die Ausführung des obigen Codes liefert folgende Ergebnisse:
Wenn in Zeile 17 der Scheduler [Schedulers.trampoline] verwendet wird, erhält man die umgekehrten Ergebnisse:
Allerdings ist es schwierig, einen Zusammenhang mit den Observables herzustellen. Ich habe kein überzeugendes Beispiel gefunden, das den Nutzen der Ausführung eines Observables auf einem dieser beiden Threads hätte verdeutlichen können. Hier ist dennoch eines, das ich jedoch überhaupt nicht natürlich finde:
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 {
// Worker
Worker worker = Schedulers.immediate().createWorker();
// Worker worker = Schedulers.trampoline().createWorker();
// Observable 1 auf Worker
worker.schedule(() -> Observable.range(1, 2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer i) {
ProcessUtils.showInfos.accept(String.valueOf(i));
// Observable 2 auf demselben Worker
worker.schedule(() -> Observable.range(100, 2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer i) {
ProcessUtils.showInfos.accept(String.valueOf(i));
}
}));
}
}));
}
}
- Zeilen 13–14: Es wird ein Worker aus einem der beiden Scheduler [Schedulers.immediate] und [Schedulers.trampoline] erstellt;
- Zeile 16: Ein erstes Observable obs1 wird auf diesem Worker geplant, um die Zahlen [1,2] auszugeben
- Zeile 22: Jedes Mal, wenn ein Element dieses Observables obs1 beobachtet wird, wird die Beobachtung eines zweiten Observables obs2 auf demselben Worker ausgelöst, um die Zahlen [100,101] auszugeben;
Mit dem Scheduler [Schedulers.immediate] erhält man folgende Ergebnisse:
Mit dem Scheduler [Schedulers.trampoline] erhält man hingegen folgende Ergebnisse:
7.8. Conclusion
Es gibt noch viel zu tun. Um sich eingehender mit der Bibliothek RxJava zu befassen, wird der Leser gebeten, seine Weiterbildung anhand der zu Beginn dieses Dokuments angegebenen Referenzen fortzusetzen. Dennoch verfügen wir über die Grundlagen, um RxJava in Swing- und Android-Umgebungen zu verwenden. Das werden wir nun zeigen.