Ich habe den folgenden Code mit RxJava Observable API:
%Vor%Meine Erwartung war, dass der Beobachtungscode, d. h. der Code innerhalb der subscribe () -Methode, parallel ausgeführt wird, nachdem ich den Berechnungsplaner spezifiziert habe. Stattdessen wird der Code noch sequenziell auf einem einzelnen Thread ausgeführt. Wie kann der Code mit RxJava api parallel ausgeführt werden?
RxJava wird oft missverstanden, wenn es um asynchrone / Multithread-Aspekte geht. Die Codierung von Multithread-Operationen ist einfach, aber das Verständnis der Abstraktion ist eine andere Sache.
Eine häufige Frage zu RxJava ist, wie Parallelisierung erreicht werden kann oder mehrere Objekte gleichzeitig von einem Observable ausgegeben werden. Natürlich bricht diese Definition den Observable Contract, der besagt, dass onNext () nacheinander und niemals gleichzeitig von mehr als einem Thread aufgerufen werden muss.
Um Parallelität zu erreichen, benötigen Sie mehrere Observables.
Dies läuft in einem einzigen Thread:
%Vor%Dies läuft in mehreren Threads:
%Vor%Code und Text kommt von diesem Blogpost.
Sie müssen subscribeOn(Schedulers.computation()) anstelle von observeOn(Schedulers.computation()) für diesen Zweck angeben.
In subscribeOn deklarieren Sie, in welchen Thread Sie Ihre Werte ausgeben werden.
In observeOn deklarieren Sie, in welchem Thread Sie arbeiten und beobachten sie.
Dies kommt immer noch in der gleichen Reihenfolge. Auch bei neuen Threads
Observable ob3 = Observable.range (1, 5);
%Vor%Ausgabe 1 ccc RxNewThreadScheduler-1 2 ccc RxNewThreadScheduler-1 3 ccc RxNewThreadScheduler-1 4 ccc RxNewThreadScheduler-1 5 ccc RxNewThreadScheduler-1
Die Verwendung von flatMap und die Angabe zum Abonnieren in Schedulers.computation() erreichen einen gemeinsamen Zugriff.
Hier ist ein praktischeres Beispiel mit Callable , aus der Ausgabe können wir sehen, dass es etwa 2000 Millisekunden dauern wird, um alle Aufgaben zu beenden.
Tags und Links java rx-java system.reactive