Unterschied zwischen Java 8-Streams und RxJava-Observablen

Question

Sind Java 8-Streams RxJava-Observablen ähnlich?

Java 8 Stream Definition:

Klassen in dem neuen Paket java.util.stream bietet einen Stream-API zu Unterstützung funktionellen Stil Operationen auf Ströme von Elementen.

Answer 1

TL; DR: Alle Sequenz/Stream-Processing-Bibliotheken bieten eine sehr ähnliche API für den Pipelinebau. Die Unterschiede bestehen in der API für die Handhabung von Multi-Threading und die Zusammensetzung von Pipelines.

RxJava ist ganz anders als Stream. Von allen JDK-Dingen ist das am nächsten zu rx.Observable vielleicht java.util.stream.Collector Stream + CompleableFuture-Combo (was zu einem Preis des Umgangs mit einer zusätzlichen Monad-Schicht führt, d.h. die Umwandlung zwischen Stream<CompletableFuture<T>> und CompletableFuture<Stream<T>> bewältigen muss).

Es gibt erhebliche Unterschiede zwischen beobachtbaren und Stream:

• Streams sind Pull-basierte, Observable sind Push-basiert. Das hört sich vielleicht zu abstrakt an, hat aber erhebliche Konsequenzen, die sehr konkret sind.
• -Stream kann nur einmal verwendet werden kann beobachtbar oft
• Stream#parallel() Splits Sequenz in Partitionen abonniert werden, Observable#subscribeOn() und Observable#observeOn() nicht; Es ist schwierig, Stream#parallel() Verhalten mit Observable zu emulieren, es hatte einmal .parallel() Methode, aber diese Methode verursacht so viel Verwirrung, dass .parallel() Unterstützung wurde auf separaten Repository auf Github, RxJavaParallel verschoben. Weitere Details sind in another answer.
• Stream#parallel() lässt nicht zu, einen Threadpool zu verwenden, anders als die meisten RxJava-Methoden, die optionalen Scheduler akzeptieren. Da alle Stream-Instanzen in einer JVM verwenden die gleiche fork-join Pool, das Hinzufügen .parallel() kann versehentlich das Verhalten in einem anderen Modul des Programms beeinflussen
• Streams fehlen zeitbezogene Operationen wie Observable#interval(), Observable#window() und viele andere; Dies liegt vor allem daran, dass die Streams Pull-basiert sind
• Streams bieten eingeschränkte Reihe von Operationen im Vergleich zu RxJava. Z.B. Die Ströme haben keine Abschneideoperationen (takeWhile(), takeUntil()); Abhilfe mit Stream#anyMatch() begrenzt: es ist Terminalbetrieb, so dass Sie nicht mehr als einmal pro Stream 8
• Ab JDK verwenden können, gibt es keinen Strom # Zip-Operation, die manchmal ganz nützlich ist
• Streams schwer um selbst zu konstruieren, Observable kann durch viele Möglichkeiten konstruiert werdenEDIT: Wie in Kommentaren erwähnt, gibt es Möglichkeiten, um Stream zu konstruieren. Da es jedoch keinen nicht-terminalen Kurzschluss gibt, können Sie nicht e. G. leicht generieren Stream von Zeilen in Datei (JDK bietet Dateien # Zeilen und BufferedReader # Linien out of the Box und andere ähnliche Szenarien können verwaltet werden, indem Stream aus Iterator).
• Observable bietet Ressourcen-Management-Einrichtung (Observable#using()); Sie können IO-Stream oder Mutex damit umhüllen und sicherstellen, dass der Benutzer nicht vergessen wird, die Ressource freizugeben - er wird automatisch bei Beendigung des Abonnements entsorgt; Stream hat onClose(Runnable) Methode, aber Sie müssen es manuell oder über Try-mit-Ressourcen aufrufen. Z.B. Sie müssen daran denken, dass Files # lines() in try-with-resources blockieren müssen.
• Observables sind vollständig synchronisiert (ich habe nicht wirklich überprüft, ob das gleiche für Streams gilt). Dies erspart Ihnen das Denken, ob grundlegende Operationen threadsicher sind (die Antwort ist immer "Ja", es sei denn, es gibt einen Fehler), aber der Nebenläufigkeits-Overhead wird da sein, egal ob Ihr Code es benötigt oder nicht.

Zusammenfassung: RxJava unterscheidet sich erheblich von Streams. Echte RxJava-Alternativen sind andere Implementierungen von ReactiveStreams, z. G. relevanter Teil von Akka.

Aktualisieren. Es gibt Trick zu verwenden Nicht-Standard-fork-join Pool für Stream#parallel finden Custom thread pool in Java 8 parallel stream

aktualisieren. All dies basiert auf den Erfahrungen mit RxJava 1.x. Jetzt, dass RxJava 2.x is here, diese Antwort möglicherweise nicht mehr aktuell ist.

Answer 2

Java 8 Stream und RxJava sieht sehr ähnlich aus. Sie haben ähnliche Operatoren (Filter, Map, FlatMap ...), sind aber nicht für die gleiche Verwendung gebaut.

Mit RxJava können Sie asynchonus-Aufgaben ausführen.

Mit Java 8 streamen Sie Elemente Ihrer Sammlung.

Sie können so ziemlich dasselbe in RxJava (traverse Elemente einer Sammlung) tun, aber, wie RxJava auf gleichzeitige Aufgabe konzentriert ist, ..., verwenden Sie Synchronisation, Latch, ... Also die gleiche Aufgabe mit RxJava möglicherweise langsamer als mit Java 8-Stream.

RxJava kann mit CompletableFuture verglichen werden, aber das kann in der Lage sein, mehr als nur einen Wert zu berechnen.

Answer 3

Es gibt einige technische und konzeptionelle Unterschiede, zum Beispiel Java 8-Streams sind single-use, Pull-basierte, synchrone Folgen von Werten, während RxJava Observables sind wiederbeobachtbare, adaptive Push-Pull-basierte, möglicherweise asynchrone Folgen von Werten. RxJava ist auf Java 6+ ausgerichtet und funktioniert auch auf Android.

Answer 4

Java 8 Streams sind Pull-basiert. Sie durchlaufen einen Java 8-Stream, der jedes Element verbraucht. Und es könnte ein endloser Strom sein.

RXJava Observable ist standardmäßig Push-basiert. Sie abonnieren ein Observable und Sie werden benachrichtigt, wenn das nächste Element eintrifft (onNext) oder wenn der Stream abgeschlossen ist (onCompleted) oder wenn ein Fehler aufgetreten ist (onError). Denn mit Observable Sie erhalten onNext, onCompleted, onError Ereignisse, können Sie wie die Kombination verschiedener Observable s auf einen neuen einige leistungsstarke Funktionen tun (zip, merge, concat). Andere Dinge, die Sie tun könnten, ist Caching, Throttling, ... Und es verwendet mehr oder weniger die gleiche API in verschiedenen Sprachen (RxJava, RX in C#, RxJS, ...)

Standardmäßig RxJava ist single-threaded. Wenn Sie keine Scheduler verwenden, geschieht alles auf demselben Thread.

Answer 5

RxJava ist auch eng mit der reactive streams initiative verbunden und betrachtet es als eine einfache Implementierung der reaktiven Ströme API (z. B. im Vergleich zu der Akka streams implementation). Der Hauptunterschied besteht darin, dass die reaktiven Ströme so ausgelegt sind, dass sie den Gegendruck beherrschen, aber wenn Sie sich die Seite mit den reaktiven Datenströmen ansehen, werden Sie auf die Idee kommen. Sie beschreiben ihre Ziele ziemlich gut und die Ströme sind auch eng verwandt mit the reactive manifesto.

Die Java 8-Streams sind so ziemlich die Implementierung einer unbegrenzten Sammlung, ziemlich ähnlich der Scala Stream oder der Clojure lazy seq.

Answer 6

Java 8-Streams ermöglichen die effiziente Verarbeitung sehr großer Sammlungen bei gleichzeitiger Nutzung von Multicore-Architekturen. Im Gegensatz dazu ist RxJava standardmäßig single-threaded (ohne Scheduler). Daher wird RxJava Multi-Core-Rechner nicht nutzen, wenn Sie diese Logik nicht selbst programmieren.

Answer 7

Die vorhandenen Antworten sind umfassend und korrekt, aber ein klares Beispiel für Anfänger fehlt. Gestatten Sie mir, einige Begriffe wie "push/pull-based" und "re-observable" zu hinterfragen. Hinweis: Ich hasse den Begriff Observable (es ist ein Strom um Himmels Willen), so wird einfach auf J8 vs RX-Streams beziehen.

Betrachten wir eine Liste, wenn ganze Zahlen,

digits = [1,2,3,4,5] 

Ein J8-Stream ist ein Dienstprogramm, um die Sammlung zu ändern. Zum Beispiel können auch Ziffern extrahiert werden,

evens = digits.stream().filter(x -> x%2) 

Diese im Grunde das ist Python map, filter, reduce, ein sehr schönes (und längst überfälligen) zusätzlich zu Java. Was aber, wenn die Ziffern nicht vorher gespeichert werden? Kann man die geraden Ziffern noch filtern?

Stellen ein separater Thread Prozess ganze Zahlen zufällig Ausgeben (--- bezeichnet die Zeit)

digits = 12345---6------7--8--9-10--------11--12 

In RX, even auf jede neue digit reagieren kann und die Filter in Echtzeit

even = -2-4-5---6---------8----10------------12 

gilt

Es ist nicht erforderlich, Eingabe- und Ausgabelisten zu speichern. Wenn Sie eine Ausgabeliste wollen, kein Problem, das auch streamable ist. In der Tat, everything is a stream.

evens_stored = even.collect() 

Aus diesem Grund Begriffe wie "stateless" und "funktional" sind mehr mit RX