Наблюдаемый против Flowable rxJava2

128

Я смотрел на новый rx java 2, и я не совсем уверен, что понимаю идею backpressure...

Я знаю, что у нас есть Observableто, у кого нет backpressureподдержки, а у Flowableтого есть.

Так на основе , например, позволяет сказать , что у меня есть flowableс interval:

        Flowable.interval(1, TimeUnit.MILLISECONDS, Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<Long>() {
                @Override
                public void accept(Long aLong) throws Exception {
                    // do smth
                }
            });

Это выйдет из строя примерно после 128 значений, и это довольно очевидно, что я потребляю медленнее, чем получаю предметы.

Но тогда у нас то же самое с Observable

     Observable.interval(1, TimeUnit.MILLISECONDS, Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<Long>() {
                @Override
                public void accept(Long aLong) throws Exception {
                    // do smth
                }
            });

Это вообще не приведет к сбою, даже если я отложу потребление, он все еще работает. Чтобы заставить Flowableработать, скажем, я поставил onBackpressureDropоператор, сбой исчез, но также не все значения выдаются.

Итак, основной вопрос, на который я не могу найти ответа в настоящее время в моей голове, заключается в том, почему я должен заботиться о том, backpressureкогда я могу использовать plain, Observableвсе равно получать все значения без управления buffer? Или, может быть, с другой стороны, какие преимущества backpressureдает мне управление и управление потреблением?

user2141889
источник

Ответы:

123

На практике противодавление проявляется в ограниченных буферах Flowable.observeOnс буфером из 128 элементов, который истощается так быстро, как это может принять нисходящий поток. Вы можете увеличивать этот размер буфера индивидуально для обработки импульсного источника, и все методы управления противодавлением по-прежнему применяются с версии 1.x. Observable.observeOnимеет неограниченный буфер, который продолжает собирать элементы, и вашему приложению может не хватить памяти.

Вы можете использовать, Observableнапример:

  • обработка событий GUI
  • работа с короткими последовательностями (всего менее 1000 элементов)

Вы можете использовать, Flowableнапример:

  • холодные и несвязанные источники
  • генератор как источники
  • аксессоры сети и базы данных
akarnokd
источник
Так как это придумал другой вопрос - это правильно , что более ограниченные типы , как Maybe, Singleи Completableможет всегда использоваться вместо того , чтобы, Flowableкогда они семантически соответствующие?
david.mihola 01
1
Да, Maybe, Singleи Completableявляются далеко слишком малы , чтобы иметь какой - либо необходимости концепции противодавления. Нет никакого шанса, что производитель может выпускать предметы быстрее, чем они могут быть потреблены, поскольку 0–1 предмет когда-либо будет произведен или потреблен.
AndrewF 02
Возможно, я не прав, но для меня следует поменять местами примеры Flowable и Observable.
Юра Галавай
Я думаю, что в вопросе ему не хватает стратегии противодавления, которую мы должны предоставить Flowable, которая объясняет, почему возникает исключение отсутствующего противодавления, а также объясняет, почему это исключение исчезает после того, как он применил .onBackpressureDrop (). А для Observable, поскольку у него нет этой стратегии и она не может быть предоставлена, он просто потерпит неудачу позже из-за OOM
Хаомин
111

Обратное давление - это когда ваш наблюдаемый (издатель) создает больше событий, чем может обработать ваш подписчик. Таким образом, вы можете заставить подписчиков пропустить события или получить огромную очередь событий, которая в конечном итоге приводит к нехватке памяти. Flowableучитывает противодавление. Observableне. Это оно.

это напоминает мне воронку, которая, когда в ней слишком много жидкости, переливается. Flowable может помочь избежать этого:

с огромным противодавлением:

введите описание изображения здесь

но при использовании текучей среды обратное давление намного меньше:

введите описание изображения здесь

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

вот стратегии. Я не буду перечислять их все, но, например, если вы не хотите беспокоиться о переполненных элементах, вы можете использовать такую ​​стратегию сброса:

observable.toFlowable (BackpressureStrategy.DROP)

Насколько я знаю, в очереди должно быть ограничение в 128 элементов, после чего может произойти переполнение (противодавление). Даже если это не 128, оно близко к этому числу. Надеюсь, это кому-то поможет.

если вам нужно изменить размер буфера со 128, похоже, это можно сделать следующим образом (но обратите внимание на любые ограничения памяти:

myObservable.toFlowable(BackpressureStrategy.MISSING).buffer(256); //but using MISSING might be slower.  

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

j2emanue
источник
Я всегда думал, что противодавление - это название семейства механизмов, которые позволят потребителю уведомить производителя о замедлении ...
kboom
Может быть. Да
j2emanue
Есть ли недостатки у использования Flowable?
IgorGanapolsky
Эти изображения лгут мне. Отбрасывание событий не приведет к тому, что внизу будет «больше денег».
EpicPandaForce
1
@ j2emanue, вы путаете размер буфера для операторов и оператора Flowable.buffer (int). Внимательно прочтите javadocs и соответствующим образом исправьте свой ответ: reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html
tomek
15

Тот факт, что у вас Flowableпроизошел сбой после выдачи 128 значений без обработки противодавления, не означает, что он всегда будет аварийно завершать работу ровно после 128 значений: иногда он вылетает после 10, а иногда вообще не падает. Я считаю, что это то, что произошло, когда вы попробовали пример с Observable- не было никакого противодавления, поэтому ваш код работал нормально, в следующий раз может не работать. Разница в RxJava 2 заключается в том, что в Observables больше нет концепции противодавления и нет способа справиться с этим. Если вы разрабатываете реактивную последовательность, которая, вероятно, потребует явной обработки противодавления - Flowableэто ваш лучший выбор.

Егор
источник
Да, я заметил, что иногда он ломался после меньших значений, иногда нет. Но опять же, если, например, я работаю только intervalбез backpressure, ожидать ли я какого-то странного поведения или проблем?
user2141889
Если вы уверены, что проблемы с противодавлением не могут возникнуть в определенной наблюдаемой последовательности - тогда, я думаю, можно игнорировать противодавление.
Егор