Reactive Extensions (Rx) sind eine Sammlung von Methoden und Schnittstellen, die schnelle, leicht zu wartende und verständliche Lösungen für die Probleme von Java developers bieten. Genau, Rx ist ein Werkzeugsatz, der Ihnen hilft, eleganten und dennoch einfachen Code zu schreiben.
Bei Java ermöglicht die Hinzufügung der Rx-Bibliothek die Orchestrierung mehrerer Aktionen, die aufgrund bestimmter Ereignisse im System auftreten. Außerdem nimmt sie die Mühe der Verwendung von Callbacks und des komplizierten globalen Zustandsmanagements weg.
Wenn Sie gerade erst anfangen, mit RxJava zu arbeiten, könnten Sie denken, dass es zu schwierig zu verstehen ist und dieser Satz neuer Werkzeuge nicht viel hilft. Doch mit der Zeit werden Sie die zahlreichen Vorteile von Rx kennenlernen:
Intuitivität — Die Beschreibung von Aktionen in Rx verwendet denselben Stil wie in der funktionalen Programmierung, zum Beispiel bei Java Streams. Wenn Sie mit Rx arbeiten, haben Sie die Möglichkeit, funktionale Transformationen über Ereignisströme hinweg zu nutzen.
Erweiterbarkeit — RxJava kann durch benutzerdefinierte Operatoren erweitert werden.
Deklarativität — funktionale Transformationen werden deklarativ beschrieben.
Komponierbarkeit — operatoren in RxJava lassen sich leicht zu komplexen Operationen zusammensetzen.
Konvertierbarkeit — operatoren in RxJava können Datentypen durch Filtern, Verarbeiten und Erweitern von Datenströmen transformieren.
Rx basiert auf dem Observer-Muster. Die Schlüsselelemente sind der Observer und der Subscriber. Observable ist der Basistyp. Diese Klasse enthält den Hauptteil der Rx-Implementierung und umfasst alle grundlegenden Operatoren.
Warum RxJava?
1. Asynchrone Streams
2. Funktionaler Ansatz
3. Einfaches Caching
4. Operatoren mit Schedulers
5. Verwendung von Subjects
1. Asynchrone Streams
Nehmen wir ein Beispiel:
Sie müssen eine Anfrage an die Datenbank senden und danach sofort mit dem Abrufen von Nachrichten und Einstellungen beginnen. Nachdem die Arbeit abgeschlossen ist, müssen Sie eine Willkommensnachricht anzeigen.
Wenn wir dasselbe in Java SE und Android tun wollen, müssen wir die folgenden Schritte ausführen:
- 3-4 verschiedene AsyncTasks ausführen;
- Ein Semaphor erstellen, der auf das Abschließen beider Anfragen wartet (Einstellungen und Nachrichten);
- Felder auf Objektebene erstellen, um die Ergebnisse zu speichern;
Wir können all diese Operationen minimieren, indem wir mit RxJava arbeiten. In diesem Fall sieht der Code wie ein Thread aus, der an einem Ort steht und auf der Grundlage einer funktionalen Paradigma aufgebaut ist.
Alternativ können wir optionales Binden verwenden:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
Observable.fromCallable(createNewUser()) .subscribeOn(Schedulers.io()) .flatMap(new Func1<User, Observable<Pair<Settings, List>>>() { // works with settings }) .doOnNext(new Action1<Pair<Settings, List>>() { @Override public void call(Pair<Settings, List> pair) { System.out.println("Received settings" + pair.first); } }) .flatMap(new Func1<Pair<Settings, List>, Observable>() { //works with messages }) .subscribe(new Action1() { @Override public void call(Message message) { System.out.println("New message " + message); } }); |
2. Funktionaler Ansatz
Wenn Sie mit der funktionalen Programmierung wie den Konzepten von map und zip, vertraut sind, wird die Arbeit mit RxJava für Sie viel einfacher. Wir können sagen, dass funktionale Programmierung die aktive Verwendung von Funktionen als Parameter und Ergebnisse in anderen Funktionen ist. Zum Beispiel ist map eine Funktion höherer Ordnung, die in vielen Programmiersprachen verwendet wird. Sie wendet diese Funktion auf jedes Element in der Liste an und gibt eine Liste von Ergebnissen zurück. So sieht es aus:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
Observable.from(jsonFile) .map(new Func1<File, String>() { @Override public String call(File file) { try { return new Gson().toJson(new FileReader(file), Object.class); } catch (FileNotFoundException e) { // this exception is a part of rx-java throw OnErrorThrowable.addValueAsLastCause(e, file); } } }); |
3. Einfaches Caching
Der nächste Codeabschnitt verwendet die caching method so, dass nur eine Kopie das Ergebnis speichert, nachdem es das erste Mal erfolgreich war.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
Single < List < Todo >> todosSingle = Single.create(emitter - > { Thread thread = new Thread(() - > { try { List < Todo > todosFromWeb = // query a webservice System.out.println("I am only called once!"); emitter.onSuccess(todosFromWeb); } catch (Exception e) { emitter.onError(e); } }); thread.start(); }); // cache the result of the single, so that the web query is only done once Single < List < Todo >> cachedSingle = todosSingle.cache(); |
4. Operatoren mit Schedulers
Es gibt eine Vielzahl von Operatoren, die einen Scheduler benötigen, um zu funktionieren. Gleichzeitig haben sie überladene Methoden, die computation() oder delay() als Scheduler verwenden. Hier ist ein Beispiel:
|
1 2 3 4 5 6 |
TestSubscriber subscriber = new TestSubscriber<>(); Observable.just(1).delay(1, TimeUnit.SECONDS).subscribe(subscriber); subscriber.awaitTerminalEvent(); Logger.d("LastSeenThread: " + subscriber.getLastSeenThread().getName()); |
Wir können das nächste Ergebnis sehen, obwohl wir keinen Scheduler angegeben haben. Das Ergebnis wird wie folgt aussehen, auch wenn wir keinen Scheduler angegeben haben:
|
1 2 3 |
LastSeenThread: RxComputationThreadPool-1 |
Wenn Sie vermeiden möchten, den computation Scheduler zu verwenden, sollten Sie die erforderlichen Scheduler als drittes Argument angeben:
|
1 2 3 |
.delay(1, TimeUnit.SECONDS, Schedulers.immediate()) |
Neben delay() gibt es viele andere Operatoren, die den Scheduler ändern können: interval (), timer (), überladene Methoden von buffer (), debounce (), skip (), take (), timeout () usw.
5. Verwendung von Subjects
Bei der Arbeit mit Objects sollten Sie berücksichtigen, dass die Sequenz der Datenänderungen, die an das onNext-Subject gesendet werden, standardmäßig im selben Thread ausgeführt wird, der zum Aufrufen der onNext() Methode verwendet wurde, bis der observeOn() Operator in dieser Sequenz erscheint.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
BehaviorSubject<object> subject = BehaviorSubject.create();subject.doOnNext(obj -> Logger.logThread("doOnNext")) .subscribeOn(Schedulers.io()) .observeOn(Schedulers .newThread()) .subscribe(new Subscriber<object>() { @Overridepublic void onNext(Object o) { Logger.logThread("onNext"); } }); subject.onNext("str"); Handler handler = new Handler(); handler.postDelayed(() -> subject.onNext("str"), 1000); |
Sowohl observeOn als auch subscribeOn werden hier erwähnt, aber das Ergebnis wird wie folgt sein:
|
1 2 3 |
doOnNext: RxCachedThreadScheduler-1onNext: RxNewThreadScheduler-1doOnNext: mainonNext: RxNewThreadScheduler-1 |
Dies bedeutet, dass, wenn wir uns auf ein subject abonnieren, es sofort den Wert zurückgibt und dann in einem Thread von Schedulers.io() verarbeitet wird. Wenn die nächste Nachricht beim Subject ankommt, verwenden wir den Thread, in dem onNext() aufgerufen wurde.
Fazit
Die Verwendung von Rx hilft uns, folgende Probleme zu lösen:
- Das Problem mit dem Caching auf einfache Weise zu lösen (ohne Caching-Klassen zu erstellen);
- Standard-AsyncTask loszuwerden, Anfragen zu kombinieren und Ergebnisse während des Empfangs zu verarbeiten;
- Speicherlecks um 90% zu reduzieren;
- Die Anzahl der Codezeilen zu reduzieren und die Reaktionszeit der Anwendung zu erhöhen. Es ist nicht erforderlich, viele verschiedene AsyncTasks zu schreiben, RxJava ermöglicht es, Methoden problemlos zu kombinieren;
Wie Sie sehen, ist RxJava ein neues und wirklich scharfes Werkzeug, um mit Netzwerk-Anfragen und Threads zu arbeiten, Antworten zu verarbeiten und Speicherlecks zu verhindern.
Über Redwerk
Seit 2005 bietet Redwerk Outsourcing-Programmierdienste für ein breites Spektrum von Branchen wie E-Commerce, Business Automation, E-Health, Media & Entertainment, E-Government, Spieleentwicklung und natürlich für Startups & Innovation. Unser engagiertes Offshore-Entwicklerteam mit seiner reichhaltigen Erfahrung findet immer die besten technischen Lösungen für Projekte jeder Größe. Redwerk ist in die Entwicklung von iOS- und benutzerdefinierten Android-Anwendungen von Grund auf engagiert. Wir respektieren die Zeit unserer Kunden und halten daher immer die Fristen ein.
