Разница между потоками Java 8 и наблюдаемыми в RxJava

144

Потоки Java 8 похожи на наблюдаемые RxJava?

Определение потока Java 8:

Классы в новом java.util.streamпакете предоставляют Stream API для поддержки операций в функциональном стиле над потоками элементов.

rahulrv
источник
8
К вашему сведению, в JDK есть предложения о введении новых классов, подобных RxJava
Джон Винт
@JohnVint Каков статус этого предложения. Будет ли это на самом деле полет?
Игорь Ганапольский
2
@IgorGanapolsky О да, похоже, он превратится в jdk9. cr.openjdk.java.net/~martin/webrevs/openjdk9/… . Есть даже порт для RxJava для потока github.com/akarnokd/RxJavaUtilConcurrentFlow .
Джон Винт
Я знаю, что это действительно старый вопрос, но недавно я присутствовал на этом замечательном выступлении Venkat Subramaniam, в котором подробно рассказывается об этой теме и он обновлен до Java9: youtube.com/watch?v=kfSSKM9y_0E . Может быть интересным для людей, изучающих RxJava.
Педро

Ответы:

152

TL; DR : Все библиотеки обработки последовательности / потока предлагают очень похожий API для построения конвейера. Различия в API для обработки многопоточности и составления конвейеров.

RxJava довольно сильно отличается от Stream. Из всех вещей JDK наиболее близким к rx.Observable является, возможно, комбо java.util.stream.Collector Stream + CompletableFuture (которое обходится ценой работы с дополнительным слоем монады, то есть с обработкой преобразования между Stream<CompletableFuture<T>>и CompletableFuture<Stream<T>>).

Существуют значительные различия между Observable и Stream:

  • Потоки основаны на подтягивании, Observables основаны на подталкивании. Это может показаться слишком абстрактным, но оно имеет значительные конкретные последствия.
  • Поток может быть использован только один раз, Observable может быть подписан много раз
  • Stream#parallel()разбивает последовательность на разделы Observable#subscribeOn()и Observable#observeOn()не делает; эмулировать Stream#parallel()поведение с Observable сложно , у него когда-то был .parallel()метод, но этот метод вызвал столько путаницы, что .parallel()поддержка была перемещена в отдельный репозиторий на github, RxJavaParallel. Подробнее в другом ответе .
  • Stream#parallel()не позволяет указывать пул потоков для использования, в отличие от большинства методов RxJava, принимающих необязательный планировщик. Поскольку все экземпляры потока в JVM используют один и тот же пул разветвления, добавление .parallel()может случайно повлиять на поведение в другом модуле вашей программы.
  • Потоки операций не хватает времени , связанных как Observable#interval(), Observable#window()и многие другие; Это происходит главным образом потому, что потоки основаны на извлечении, и восходящий поток не имеет никакого контроля над тем, когда испускать следующий элемент ниже по потоку.
  • Потоки предлагают ограниченный набор операций по сравнению с RxJava. Например, в потоках отсутствуют операции отключения ( takeWhile(), takeUntil()); использование обходного пути Stream#anyMatch()ограничено: это терминальная операция, поэтому вы не можете использовать его более одного раза для каждого потока
  • Начиная с JDK 8 нет операции Stream # zip, которая иногда бывает весьма полезна
  • Потоки трудно построить самостоятельно, Observable может быть построена многими способами EDIT: Как было отмечено в комментариях, есть способы построения потока. Однако, поскольку нет терминального короткого замыкания, вы, например, не можете легко сгенерировать Поток строк в файле (хотя JDK предоставляет строки Files # lines и BufferedReader # прямо из коробки, и другими подобными сценариями можно управлять, создавая Stream от итератора).
  • Наблюдаемые предложения объекта управления ресурсами ( Observable#using()); вы можете обернуть его потоком ввода-вывода или мьютексом и быть уверенным, что пользователь не забудет освободить ресурс - он будет автоматически удален при прекращении подписки; У Stream есть onClose(Runnable)метод, но вы должны вызывать его вручную или через try-with-resources. Например Вы должны иметь в виду, что Files # lines () должен быть заключен в блок try-with-resources.
  • Observables синхронизируются на всем протяжении (я фактически не проверял, верно ли то же самое для потоков). Это избавляет вас от размышлений о том, являются ли базовые операции поточно-ориентированными (ответ всегда «да», если нет ошибки), но накладные расходы, связанные с параллелизмом, будут присутствовать, независимо от того, нужен ваш код или нет.

Обзор: RxJava значительно отличается от Streams. Реальные альтернативы RxJava - это другие реализации ReactiveStream , например, соответствующая часть Akka.

Обновление . Есть хитрость для использования пула разветвления не по умолчанию Stream#parallel, см. Пользовательский пул потоков в параллельном потоке Java 8

Обновление . Все вышеперечисленное основано на опыте работы с RxJava 1.x. Теперь, когда RxJava 2.x уже здесь , этот ответ может быть устаревшим.

Кирилл Гамазков
источник
2
Почему потоки трудно построить? Согласно этой статье, это кажется простым: oracle.com/technetwork/articles/java/…
IgorGanapolsky
2
Существует целый ряд классов, у которых есть метод stream: коллекции, входные потоки, файлы каталогов и т. Д. Но что, если вы хотите создать поток из пользовательского цикла - скажем, итерации по курсору базы данных? Наилучший способ, который я нашел до сих пор, - это создать Iterator, обернуть его Spliterator и, наконец, вызвать StreamSupport # fromSpliterator. Слишком много клея для простого случая ИМХО. Существует также Stream.iterate, но он создает бесконечный поток. Единственный способ отсечь поток в этом случае - это Stream # anyMatch, но это терминальная операция, поэтому вы не можете разделить производителя и потребителя потока
Кирилл Гамазков
2
RxJava имеет Observable.fromCallable, Observable.create и так далее. Или вы можете безопасно производить бесконечное Observable, а затем сказать «.takeWhile (условие)», и вы согласны с доставкой этой последовательности потребителям
Кирилл Гамазков
1
Потоки не сложно построить самостоятельно. Вы можете просто вызвать Stream.generate()и передать свою собственную Supplier<U>реализацию, всего лишь один простой метод, из которого вы предоставляете следующий элемент в потоке. Есть множество других методов. Чтобы легко построить последовательность, Streamкоторая зависит от предыдущих значений, вы можете использовать interate()метод, каждый из которых Collectionимеет stream()метод и Stream.of()создает Streamиз varargs или массива. Наконец, StreamSupportесть поддержка более продвинутого создания потоков с использованием сплитераторов или потоков примитивных типов.
JBX
«В потоках отсутствуют операции отключения ( takeWhile(), takeUntil());» - У JDK9 есть эти, я полагаю, функции takeWhile () и dropWhile ()
Абдул
50

Java 8 Stream и RxJava выглядят очень похоже. Они имеют похожие операторы (filter, map, flatMap ...), но не созданы для одного и того же использования.

Вы можете выполнять асинхронные задачи, используя RxJava.

С потоком Java 8 вы будете проходить элементы вашей коллекции.

Вы можете сделать почти то же самое в RxJava (обход элементов коллекции), но, поскольку RxJava сфокусирован на параллельной задаче, ..., он использует синхронизацию, защелку, ... Так что та же задача с использованием RxJava может быть медленнее, чем с потоком Java 8.

С RxJava можно сравнить CompletableFuture, но он может вычислять более одного значения.

dwursteisen
источник
12
Стоит отметить, что утверждение о прохождении потока верно только для непараллельного потока. parallelStreamподдерживает аналогичную синхронизацию простых прохождений / карт / фильтрации и т. д.
Джон Винт
2
Я не думаю, что «поэтому та же задача с использованием RxJava может быть медленнее, чем с потоком Java 8». справедливо повсеместно, в значительной степени зависит от поставленной задачи.
Дашл
1
Я рад, что вы сказали, что та же задача с использованием RxJava может быть медленнее, чем с потоком Java 8 . Это очень важное различие, о котором многие пользователи RxJava не знают.
Игорь Ганапольский
RxJava является синхронным по умолчанию. Есть ли у вас какие-либо контрольные показатели в поддержку вашего заявления о том, что оно может быть медленнее?
Марцин Козиньски,
6
@ marcin-koziński, вы можете проверить этот тест: twitter.com/akarnokd/status/752465265091309568
dwursteisen
37

Есть несколько технических и концептуальных различий, например, потоки Java 8 являются одноразовыми, основанными на извлечении, синхронными последовательностями значений, тогда как Наблюдаемые RxJava являются повторно наблюдаемыми, адаптивно основанными на двухтактных, потенциально асинхронных последовательностях значений. RxJava нацелен на Java 6+ и работает на Android.

akarnokd
источник
4
Типичный код, использующий RxJava, интенсивно использует лямбды, которые доступны только в Java 8 и далее. Так что вы можете использовать Rx с Java 6, но код будет шумным
Кирилл Гамазков
1
Аналогичное различие заключается в том, что наблюдаемые Rx могут оставаться в живых до тех пор, пока не будут отписаны. Потоки Java 8 по умолчанию завершаются операциями.
Игорь Ганапольский
2
@KirillGamazkov вы можете использовать retrolambda, чтобы сделать ваш код красивее при нацеливании на Java 6.
Marcin Koziński
Котлин выглядит даже сексуальнее, чем дооснащение
Кирилл Гамазков
30

Потоки Java 8 основаны на потоке. Вы перебираете поток Java 8, потребляющий каждый элемент. И это может быть бесконечный поток.

RXJava Observableпо умолчанию основан на push. Вы подписываетесь на Observable, и вы получите уведомление, когда прибывает следующий элемент ( onNext), или когда поток завершен ( onCompleted), или когда произошла ошибка (onError ). Потому что с Observableвы получаете onNext, onCompleted, onErrorсобытия, вы можете сделать некоторые мощные функции , такие как объединение различных Observableсек на новый ( zip, merge, concat). Другие вещи, которые вы могли бы сделать, это кэширование, регулирование ... И он использует более или менее один и тот же API на разных языках (RxJava, RX в C #, RxJS, ...)

По умолчанию RxJava является однопоточным. Если вы не начнете использовать Планировщики, все будет происходить в одном потоке.

Барт Де Нойтер
источник
в Stream у вас есть forEach, это почти то же самое, что onNext
Пол
На самом деле потоки обычно являются терминальными. «Операции, которые закрывают потоковый конвейер, называются терминальными операциями. Они генерируют результат из конвейера, такого как List, Integer или даже void (любой тип, отличный от Stream)». ~ oracle.com/technetwork/articles/java/…
Игорь Ганапольский,
26

Существующие ответы являются исчерпывающими и правильными, но для начинающих отсутствует четкий пример. Позвольте мне поставить некоторые конкретные термины, такие как «push / pull-based» и «re-наблюдаемый». Примечание : я ненавижу терминObservable (ради всего святого), поэтому просто буду ссылаться на потоки J8 и RX.

Рассмотрим список целых чисел,

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

J8 Stream - это утилита для изменения коллекции. Например, даже цифры могут быть извлечены как,

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

Это в основном карта Python , фильтрация, уменьшение , очень хорошее (и давно назревшее) дополнение к Java. Но что, если цифры не были собраны раньше времени - что, если цифры были переданы во время работы приложения - мы могли бы отфильтровать четные числа в режиме реального времени.

Представьте, что отдельный процесс потока выводит целые числа в случайное время, пока приложение работает ( ---обозначает время)

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

В RX evenможет реагировать на каждую новую цифру и применять фильтр в режиме реального времени

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

Нет необходимости хранить списки ввода и вывода. если ты хотите получить выходной список, нет проблем, это тоже можно стримировать. На самом деле все это поток.

evens_stored = even.collect()  

Вот почему такие термины, как «без состояний» и «функционал», больше связаны с RX

Адам Хьюз
источник
Но 5 даже не ... И это выглядит так, как будто J8 Stream является синхронным, а Rx Stream - асинхронным?
Франклин Ю
1
@FranklinYu спасибо, я исправил 5 опечаток. Если думать меньше с точки зрения синхронных против асинхронных, хотя это может быть правильным, и больше с точки зрения императивных и функциональных. В J8 вы сначала собираете все свои предметы, а затем применяете фильтр. В RX вы определяете функцию фильтра, независимую от данных, а затем связываете ее с четным источником (живой поток или коллекция Java) ... это совершенно другая модель программирования
Адам Хьюз
Я очень удивлен этим. Я уверен, что потоки Java могут быть сделаны из потоков данных. Что заставляет вас думать об обратном?
Vic Seedoubleyew
4

RxJava также тесно связан с инициативой реактивных потоков и считает себя простой реализацией API реактивных потоков (например, по сравнению с реализацией потоков Akka ). Основное отличие состоит в том, что реактивные потоки спроектированы так, чтобы они могли справляться с обратным давлением, но если вы заглянете на страницу реактивных потоков, вы поймете идею. Они очень хорошо описывают свои цели, и потоки также тесно связаны с реактивным манифестом .

Потоки Java 8 в значительной степени являются реализацией неограниченной коллекции, очень похожей на Scala Stream или Clojure lazy seq .

Никлас Мейер
источник
3

Java 8 Streams позволяет эффективно обрабатывать действительно большие коллекции, используя многоядерные архитектуры. Напротив, RxJava является однопоточным по умолчанию (без планировщиков). Поэтому RxJava не будет использовать преимущества многоядерных машин, если вы сами не создадите эту логику.

IgorGanapolsky
источник
4
Поток также является однопоточным по умолчанию, если только вы не вызываете .parallel (). Кроме того, Rx дает больше контроля над параллелизмом.
Кирилл Гамазков
@KirillGamazkov Поток Kotlin Coroutines (на основе потоков Java8) теперь поддерживает структурированный параллелизм: kotlinlang.org/docs/reference/coroutines/flow.html#flows
IgorGanapolsky
Правда, но я ничего не сказал о Flow и структурированном параллелизме. Мои две точки зрения были следующими: 1) оба потока и Rx являются однопоточными, если вы не изменили это явно; 2) Rx дает вам детальный контроль над тем, какой шаг выполнять с каким пулом потоков, в отличие от только потоков, позволяя вам сказать «сделайте это как-нибудь параллельно»
Кирилл Гамазков
Я не совсем понимаю вопрос "для чего вам нужен пул потоков". Как вы сказали, «для эффективной обработки действительно больших коллекций». Или, может быть, я хочу, чтобы связанная с IO часть задачи выполнялась в отдельном пуле потоков. Я не думаю, что понял смысл вашего вопроса. Попробуй еще раз?
Кирилл Гамазков
1
Статические методы в классе Schedulers позволяют получать предопределенные пулы потоков, а также создавать их из Executor. См reactivex.io/RxJava/2.x/javadoc/io/reactivex/schedulers/...
Кирилл Gamazkov