Différence entre les flux Java 8 et les observables RxJava

Les flux Java 8 sont-ils similaires aux observables RxJava?

Définition du flux Java 8:

Les classes du nouveau java.util.streampackage fournissent une API Stream pour prendre en charge les opérations de style fonctionnel sur les flux d'éléments.

TL; DR : Toutes les bibliothèques de traitement de séquence / flux offrent une API très similaire pour la construction de pipelines. Les différences résident dans l'API pour la gestion du multi-threading et la composition des pipelines.

RxJava est assez différent de Stream. De toutes les choses JDK, le plus proche de rx.Observable est peut-être java.util.stream.Collector Stream + CompletableFuture combo (qui a un coût de traiter une couche monade supplémentaire, c'est-à-dire d'avoir à gérer la conversion entreStream<CompletableFuture<T>> et CompletableFuture<Stream<T>>).

Il existe des différences significatives entre Observable et Stream:

• Les flux sont basés sur l'extraction, les observables sont basés sur le push. Cela peut sembler trop abstrait, mais cela a des conséquences importantes qui sont très concrètes.
• Stream ne peut être utilisé qu'une seule fois, Observable peut être abonné plusieurs fois
• Stream#parallel()divise la séquence en partitions, Observable#subscribeOn()et Observable#observeOn()ne le fait pas; il est difficile d'émuler le Stream#parallel()comportement avec Observable, il y avait autrefois une .parallel()méthode mais cette méthode a causé tellement de confusion que le .parallel()support a été déplacé vers un référentiel séparé sur github, RxJavaParallel. Plus de détails sont dans une autre réponse .
• Stream#parallel()ne permet pas de spécifier un pool de threads à utiliser, contrairement à la plupart des méthodes RxJava acceptant le Scheduler optionnel. Étant donné que toutes les instances de flux dans une machine virtuelle Java utilisent le même pool de fork-join, l'ajout .parallel()peut affecter accidentellement le comportement dans un autre module de votre programme
• Les flux manquent d'opérations liées au temps comme Observable#interval(), Observable#window()et bien d'autres; ceci est principalement dû au fait que les flux sont basés sur l'extraction et que l'amont n'a aucun contrôle sur le moment d'émettre l'élément suivant en aval
• Les flux offrent un ensemble restreint d'opérations par rapport à RxJava. Par exemple, les flux manquent d'opérations de coupure ( takeWhile(), takeUntil()); la solution de contournement Stream#anyMatch()est limitée: il s'agit d'une opération de terminal, vous ne pouvez donc pas l'utiliser plus d'une fois par flux
• À partir de JDK 8, il n'y a pas d'opération Stream # zip, ce qui est parfois très utile
• Les flux sont difficiles à construire par vous-même, Observable peut être construit de plusieurs manières EDIT: Comme indiqué dans les commentaires, il existe des moyens de construire Stream. Cependant, comme il n'y a pas de court-circuit non terminal, vous ne pouvez pas, par exemple, générer facilement un flux de lignes dans un fichier (JDK fournit des lignes Files # et BufferedReader # prêtes à l'emploi, et d'autres scénarios similaires peuvent être gérés en construisant Stream de Iterator).
• Observable offre une fonction de gestion des ressources ( Observable#using()); vous pouvez envelopper le flux IO ou le mutex avec et être sûr que l'utilisateur n'oubliera pas de libérer la ressource - elle sera supprimée automatiquement à la résiliation de l'abonnement; Stream a une onClose(Runnable)méthode, mais vous devez l'appeler manuellement ou via try-with-resources. Par exemple. vous devez garder à l'esprit que Files # lines () doit être inclus dans le bloc try-with-resources.
• Les observables sont synchronisés tout au long (je n'ai pas vérifié si la même chose est vraie pour les flux). Cela vous évite de vous demander si les opérations de base sont sûres pour les threads (la réponse est toujours `` oui '', sauf s'il y a un bogue), mais la surcharge liée à la concurrence sera là, peu importe si votre code en a besoin ou non.

Résumé: RxJava diffère considérablement des Streams. Les alternatives réelles à RxJava sont d'autres implémentations de ReactiveStreams , par exemple une partie pertinente d'Akka.

Mettre à jour . Il existe une astuce pour utiliser un pool de fork-join non par défaut pour Stream#parallel, voir Pool de threads personnalisé dans le flux parallèle Java 8

Mettre à jour . Tout ce qui précède est basé sur l'expérience avec RxJava 1.x. Maintenant que RxJava 2.x est ici , cette réponse est peut-être obsolète.

Java 8 Stream et RxJava sont assez similaires. Ils ont des opérateurs qui se ressemblent (filter, map, flatMap ...) mais ne sont pas construits pour le même usage.

Vous pouvez effectuer des tâches asynchrones en utilisant RxJava.

Avec le flux Java 8, vous parcourez les éléments de votre collection.

Vous pouvez faire à peu près la même chose dans RxJava (traverser les éléments d'une collection) mais, comme RxJava se concentre sur la tâche simultanée, ..., il utilise la synchronisation, le verrouillage, ... Donc la même tâche utilisant RxJava peut être plus lente que avec le flux Java 8.

RxJava peut être comparé à CompletableFuture, mais cela peut être capable de calculer plus d'une valeur.

Il existe quelques différences techniques et conceptuelles, par exemple, les flux Java 8 sont des séquences de valeurs synchrones à usage unique, basées sur l'extraction, tandis que les observables RxJava sont des séquences de valeurs ré-observables, basées sur le push-pull, potentiellement asynchrones. RxJava est destiné à Java 6+ et fonctionne également sur Android.

Les flux Java 8 sont basés sur l'extraction. Vous itérez sur un flux Java 8 consommant chaque élément. Et cela pourrait être un flux sans fin.

RXJava Observableest par défaut basé sur le push. Vous vous abonnez à un Observable et vous serez averti lorsque le prochain élément arrivera ( onNext), ou lorsque le flux est terminé ( onCompleted), ou lorsqu'une erreur s'est produite ( onError). Parce qu'avec Observablevous recevez onNext, onCompleted, les onErrorévénements, vous pouvez faire quelques fonctions puissantes comme la combinaison de différentes Observables à un nouveau ( zip, merge,concat ). Vous pouvez également faire de la mise en cache, de la limitation, ... et il utilise plus ou moins la même API dans différentes langues (RxJava, RX en C #, RxJS, ...)

Par défaut, RxJava est à thread unique. À moins que vous ne commenciez à utiliser les planificateurs, tout se passera sur le même fil.

Les réponses existantes sont complètes et correctes, mais il manque un exemple clair pour les débutants. Permettez-moi de mettre du concret derrière des termes tels que «basé sur le push / pull» et «ré-observable». Remarque : je déteste le termeObservable (c'est un flux pour l'amour du ciel), je ferai donc simplement référence aux flux J8 vs RX.

Considérons une liste d'entiers,

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

Un J8 Stream est un utilitaire pour modifier la collection. Par exemple, même les chiffres peuvent être extraits comme suit:

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

Il s'agit essentiellement de la carte de Python , filtrer, réduire , un très bel ajout (et attendu depuis longtemps) à Java. Mais que se passerait-il si les chiffres n'étaient pas collectés à l'avance - et si les chiffres étaient diffusés pendant que l'application était en cours d'exécution - pourrions-nous filtrer les paires en temps réel.

Imaginez qu'un processus de thread distinct génère des nombres entiers à des moments aléatoires pendant que l'application est en cours d'exécution ( ---indique le temps)

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

Dans RX, evenpeut réagir à chaque nouveau chiffre et appliquer le filtre en temps réel

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

Il n'est pas nécessaire de stocker des listes d'entrée et de sortie. Si vous voulez une liste de sortie, pas de problème qui soit également diffusable. En fait, tout est un flux.

evens_stored = even.collect()  

C'est pourquoi des termes comme «sans état» et «fonctionnel» sont davantage associés à RX

RxJava est également étroitement lié à l' initiative de flux réactifs et se considère comme une simple implémentation de l'API de flux réactifs (par exemple par rapport à l' implémentation de flux Akka ). La principale différence est que les flux réactifs sont conçus pour pouvoir gérer la contre-pression, mais si vous regardez la page des flux réactifs, vous aurez l'idée. Ils décrivent assez bien leurs objectifs et les flux sont également étroitement liés au manifeste réactif .

Les flux Java 8 sont à peu près l'implémentation d'une collection illimitée, assez similaire au Scala Stream ou au Clojure paresseux seq .

Les flux Java 8 permettent de traiter efficacement de très grandes collections, tout en tirant parti des architectures multicœurs. En revanche, RxJava est monothread par défaut (sans Schedulers). Ainsi, RxJava ne profitera pas des machines multicœurs à moins que vous ne codiez vous-même cette logique.