Anwendungsfälle für RxJava-Scheduler

253

In RxJava stehen 5 verschiedene Scheduler zur Auswahl:

instant () : Erstellt einen Scheduler und gibt ihn zurück, der die Arbeit am aktuellen Thread sofort ausführt.

trampoline () : Erstellt einen Scheduler und gibt ihn zurück, der die Arbeit an dem aktuellen Thread in die Warteschlange stellt, der nach Abschluss der aktuellen Arbeit ausgeführt werden soll.

newThread () : Erstellt einen Scheduler und gibt ihn zurück, der für jede Arbeitseinheit einen neuen Thread erstellt.

computation () : Erstellt einen Scheduler für die Rechenarbeit und gibt ihn zurück. Dies kann für Ereignisschleifen, die Verarbeitung von Rückrufen und andere Rechenarbeiten verwendet werden. Führen Sie keine E / A-gebundenen Arbeiten an diesem Scheduler durch. Verwenden Sie Scheduler. io () stattdessen.

io () : Erstellt einen Scheduler für E / A-gebundene Arbeiten und gibt ihn zurück. Die Implementierung wird durch einen Executor-Thread-Pool unterstützt, der nach Bedarf erweitert wird. Dies kann zum asynchronen Ausführen blockierender E / A verwendet werden. Führen Sie keine Rechenarbeiten an diesem Scheduler durch. Verwenden Sie Scheduler. stattdessen computation () .

Fragen:

Die ersten drei Scheduler sind ziemlich selbsterklärend. Ich bin jedoch ein wenig verwirrt über Berechnung und Io .

Was genau ist "IO-gebundene Arbeit"? Wird es für den Umgang mit Streams ( java.io) und Dateien ( java.nio.files) verwendet? Wird es für Datenbankabfragen verwendet? Wird es zum Herunterladen von Dateien oder zum Zugreifen auf REST-APIs verwendet?
Wie unterscheidet sich computation () von newThread () ? Befinden sich alle computation () -Aufrufe jedes Mal in einem einzelnen (Hintergrund-) Thread anstelle eines neuen (Hintergrund-) Threads?
Warum ist es schlecht, bei IO-Arbeiten computation () aufzurufen ?
Warum ist es schlecht, io () aufzurufen, wenn Sie Computerarbeiten ausführen?

java multithreading thread-safety rx-java rx-android bcorso
quelle

Antworten:

332

Tolle Fragen, ich denke, die Dokumentation könnte etwas detaillierter sein.

io()wird von einem unbegrenzten Thread-Pool unterstützt und ist die Art von Dingen, die Sie für nicht rechenintensive Aufgaben verwenden würden, dh Dinge, die die CPU nicht stark belasten. Ja, die Interaktion mit dem Dateisystem, die Interaktion mit Datenbanken oder Diensten auf einem anderen Host sind gute Beispiele.
computation()wird durch einen begrenzten Thread-Pool mit einer Größe unterstützt, die der Anzahl der verfügbaren Prozessoren entspricht. Wenn Sie versucht haben, CPU-intensive Arbeit parallel auf mehr als den verfügbaren Prozessoren zu planen (z. B. mithilfe von newThread()), müssen Sie den Aufwand für die Thread-Erstellung und den Kontextwechsel auf den Kontext übertragen, da Threads um einen Prozessor wetteifern, und dies ist möglicherweise ein großer Leistungseinbruch.
Es ist am besten, nur computation()für CPU-intensive Arbeit zu gehen, sonst erhalten Sie keine gute CPU-Auslastung.
Es ist io()aus dem in 2. beschriebenen Grund schlecht, Rechenarbeit zu fordern. io()Wenn Sie tausend Rechenaufgaben io()parallel planen, hat jede dieser tausend Aufgaben jeweils einen eigenen Thread und konkurriert um die CPU, die Kosten für den Kontextwechsel verursacht.

Dave Moten
quelle

Durch Vertrautheit mit der RxJava-Quelle. Es war für mich lange Zeit eine Quelle der Verwirrung, und ich denke, die Dokumentation muss in dieser Hinsicht verbessert werden.

Dave Moten

@IgorGanapolsky Ich vermute, das ist etwas, was Sie selten tun möchten. Das Erstellen eines neuen Fadens für jede Arbeitseinheit trägt selten zur Effizienz bei, da das Konstruieren und Abreißen von Fäden teuer ist. Normalerweise möchten Sie Threads wiederverwenden, die computation () und andere Scheduler ausführen. Das einzige Mal, dass newThread () eine legitime Verwendung hat (zumindest kann ich mir vorstellen), ist das Starten von isolierten, seltenen und lang laufenden Aufgaben. Selbst dann könnte ich io () für dieses Szenario verwenden.

tmn

Können Sie ein Beispiel zeigen, in dem Trampolin () nützlich wäre? Ich verstehe das Konzept, kann aber kein Szenario herausfinden, in dem ich es in der Praxis anwenden würde. Es ist der einzige

Planer

Verwenden Sie für Netzwerkanrufe Schedulers.io () und verwenden Sie Scheduler.from (Executors.newFixedThreadPool (n)), wenn Sie die Anzahl gleichzeitiger Netzwerkanrufe begrenzen müssen.

Dave Moten

Sie könnten denken, dass das timeoutstandardmäßige Anlegen computation()einen Thread blockieren würde, aber das ist nicht der Fall. Unter der Decke computation()verwendet eine ScheduledExecutorServiceso zeitverzögerte Aktion nicht blockieren. Angesichts dieser Tatsache computation()ist dies eine gute Idee, denn wenn es sich um einen anderen Thread handelt, fallen für uns Kosten für den Threadwechsel an.

Dave Moten

Der wichtigste Punkt ist, dass sowohl Schedulers.io als auch Schedulers.computation von unbegrenzten Thread-Pools unterstützt werden, im Gegensatz zu den anderen in der Frage genannten. Diese Eigenschaft wird nur von Schedulers.from (Executor) gemeinsam genutzt, wenn der Executor mit newCachedThreadPool erstellt wurde (unbegrenzt mit einem Thread-Pool für die automatische Rückforderung).

Wie in früheren Antworten und mehreren Artikeln im Internet ausführlich erläutert, müssen Schedulers.io und Schedulers.computation sorgfältig verwendet werden, da sie für die Art der Arbeit in ihrem Namen optimiert sind. Meiner Ansicht nach besteht ihre wichtigste Aufgabe darin , reaktiven Streams echte Parallelität zu bieten .

Entgegen der Überzeugung von Neulingen sind reaktive Streams nicht von Natur aus gleichzeitig, sondern von Natur aus asynchron und sequentiell. Aus diesem Grund darf Schedulers.io nur verwendet werden, wenn die E / A-Operation blockiert ist (z. B. wenn ein Blockierungsbefehl wie Apache IOUtils FileUtils.readFileAsString (...) verwendet wird ), wodurch der aufrufende Thread eingefroren wird, bis die Operation ausgeführt wird getan.

Die Verwendung einer asynchronen Methode wie Java AsynchronousFileChannel (...) würde den aufrufenden Thread während des Vorgangs nicht blockieren, sodass die Verwendung eines separaten Threads keinen Sinn macht. Tatsächlich eignen sich Schedulers.io- Threads nicht wirklich für asynchrone Vorgänge, da sie keine Ereignisschleife ausführen und der Rückruf niemals ... aufgerufen werden würde.

Die gleiche Logik gilt für Datenbankzugriff oder Remote-API-Aufrufe. Verwenden Sie Schedulers.io nicht, wenn Sie eine asynchrone oder reaktive API zum Aufrufen verwenden können.

Zurück zur Parallelität. Möglicherweise haben Sie keinen Zugriff auf eine asynchrone oder reaktive API, um E / A-Vorgänge asynchron oder gleichzeitig auszuführen. Daher besteht Ihre einzige Alternative darin, mehrere Aufrufe an einen separaten Thread zu senden. Leider sind reaktive Streams an ihren Enden sequentiell, aber die gute Nachricht ist, dass der Operator flatMap () im Kern Parallelität einführen kann .

Parallelität muss im Stream-Konstrukt erstellt werden, normalerweise mit dem Operator flatMap () . Dieser leistungsstarke Operator kann so konfiguriert werden, dass er intern einen Multithread-Kontext für Ihre in flatMap () eingebettete Funktion <T, R> bereitstellt . Dieser Kontext wird von einem Multithread-Scheduler wie Scheduler.io oder Scheduler.computation bereitgestellt .

Weitere Informationen finden Sie in Artikeln zu RxJava2- Schedulern und Parallelität. Dort finden Sie ein Codebeispiel und detaillierte Erklärungen zur sequentiellen und gleichzeitigen Verwendung von Schedulern.

Hoffe das hilft,

Softjake

Softjake
quelle

Dieser Blog-Beitrag bietet eine hervorragende Antwort

Aus dem Blogbeitrag:

Schedulers.io () wird von einem unbegrenzten Thread-Pool unterstützt. Es wird für nicht CPU-intensive E / A-Arbeiten verwendet, einschließlich Interaktion mit dem Dateisystem, Durchführen von Netzwerkaufrufen, Datenbankinteraktionen usw. Dieser Thread-Pool soll zum asynchronen Ausführen blockierender E / A verwendet werden.

Schedulers.computation () wird von einem begrenzten Thread-Pool mit einer Größe bis zur Anzahl der verfügbaren Prozessoren unterstützt. Es wird für rechen- oder CPU-intensive Arbeiten verwendet, z. B. zum Ändern der Größe von Bildern, Verarbeiten großer Datenmengen usw. Seien Sie vorsichtig: Wenn Sie mehr Berechnungsthreads als verfügbare Kerne zuweisen, wird die Leistung aufgrund von Kontextwechsel und Thread-Erstellungsaufwand beeinträchtigt, um den sich die Threads bemühen Prozessorenzeit.

Schedulers.newThread () erstellt für jede geplante Arbeitseinheit einen neuen Thread. Dieser Scheduler ist teuer, da jedes Mal ein neuer Thread erzeugt wird und keine Wiederverwendung erfolgt.

Schedulers.from (Executor Executor) erstellt und gibt einen benutzerdefinierten Scheduler zurück, der vom angegebenen Executor unterstützt wird. Verwenden Sie Scheduler.from (Executors.newFixedThreadPool (n)), um die Anzahl gleichzeitiger Threads im Thread-Pool zu begrenzen. Dies garantiert, dass eine Aufgabe, die geplant ist, wenn alle Threads belegt sind, in die Warteschlange gestellt wird. Die Threads im Pool bleiben bestehen, bis sie explizit heruntergefahren werden.

Der Hauptthread oder AndroidSchedulers.mainThread () wird von der RxAndroid-Erweiterungsbibliothek für RxJava bereitgestellt. Im Hauptthread (auch als UI-Thread bezeichnet) findet eine Benutzerinteraktion statt. Es sollte darauf geachtet werden, diesen Thread nicht zu überladen, um zu verhindern, dass die Benutzeroberfläche nicht reagiert oder, schlimmer noch, der ANR-Dialog (Application Not Responding).

Schedulers.single () ist neu in RxJava 2. Dieser Scheduler wird von einem einzelnen Thread unterstützt, der Aufgaben nacheinander in der angeforderten Reihenfolge ausführt.

Schedulers.trampoline () führt Aufgaben auf FIFO-Weise (First In, First Out) durch einen der teilnehmenden Worker-Threads aus. Es wird häufig bei der Implementierung von Rekursion verwendet, um eine Vergrößerung des Aufrufstapels zu vermeiden.

Joe
quelle