Observable vs Flowable rxJava2

127

Ich habe mir das neue RX Java 2 angesehen und bin mir nicht ganz sicher, ob ich die Idee von verstehe backpressure mehr ...

Mir ist bewusst, dass wir Observabledas nicht backpressureunterstützen und habenFlowable das hat es.

Nehmen wir also anhand eines Beispiels an, ich habe flowablemit 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
                }
            });

Dies wird nach ungefähr 128 Werten abstürzen, und das ist ziemlich offensichtlich, dass ich langsamer konsumiere als Gegenstände zu bekommen.

Aber dann haben wir das gleiche mit 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
                }
            });

Dies wird überhaupt nicht abstürzen, selbst wenn ich den Verbrauch etwas verzögere, funktioniert es immer noch. Um die FlowableArbeit kann sagen , Ich habe onBackpressureDropBediener, crash ist weg , aber nicht alle Werte sind entweder emittiert.

Die Grundfrage, die ich derzeit nicht in meinem Kopf beantworten kann, ist, warum ich mich darum kümmern sollte, backpressurewenn ich einfach verwenden kann und Observabletrotzdem alle Werte erhalte, ohne die zu verwalten buffer. Oder vielleicht von der anderen Seite, welche Vorteile bieten backpressuremir für die Verwaltung und den Umgang mit dem Konsum?

user2141889
quelle

Antworten:

122

Was sich in der Praxis im Gegendruck manifestiert, sind begrenzte Puffer Flowable.observeOnmit einem Puffer von 128 Elementen, der so schnell entleert wird, wie es der Dowstream aufnehmen kann. Sie können diese Puffergröße einzeln erhöhen, um die Bursty-Quelle zu verarbeiten, und alle Verfahren zur Verwaltung des Gegendrucks gelten weiterhin ab 1.x.Observable.observeOnverfügt über einen unbegrenzten Puffer, der die Elemente ständig sammelt, und Ihre App verfügt möglicherweise nicht über genügend Speicher.

Sie können Observablezum Beispiel verwenden:

  • Behandlung von GUI-Ereignissen
  • Arbeiten mit kurzen Sequenzen (insgesamt weniger als 1000 Elemente)

Sie können Flowablezum Beispiel verwenden:

  • kalte und nicht zeitgesteuerte Quellen
  • Generator wie Quellen
  • Netzwerk- und Datenbankzugriffe
akarnokd
quelle
Da dies hat kommt in einer anderen Frage - ist es , dass mehr eingeschränkt Arten korrigieren wie Maybe, Singleund Completablekann immer statt verwendet werden , Flowablewenn sie semantisch angemessen sind?
David.mihola
1
Ja, Maybe, Single, und Completablesind weit zu klein jede Notwendigkeit des Rückstau Konzept zu haben. Es gibt keine Chance, dass ein Produzent Artikel schneller emittieren kann als sie konsumiert werden können, da 0–1 Artikel jemals produziert oder konsumiert werden.
AndrewF
Vielleicht habe ich nicht Recht, aber für mich sollten Beispiele für Flowable und Observable ausgetauscht werden.
Yura Galavay
Ich denke, in der Frage fehlt ihm die Gegendruckstrategie, die wir dem Flowable zur Verfügung stellen müssen. Dies erklärt, warum die fehlende Gegendruckausnahme ausgelöst wird und warum diese Ausnahme verschwindet, nachdem er .onBackpressureDrop () angewendet hat. Und für Observable wird es, da es diese Strategie nicht hat und nicht bereitgestellt werden kann, später einfach aufgrund von OOM
Haomin
110

Gegendruck tritt auf, wenn Ihr Observable (Publisher) mehr Ereignisse erstellt, als Ihr Abonnent verarbeiten kann. So können Abonnenten Ereignisse verpassen oder eine große Warteschlange von Ereignissen erhalten, die letztendlich zu Speichermangel führen. Flowableberücksichtigt den Gegendruck. Observablenicht. Das ist es.

es erinnert mich an einen Trichter, der überläuft, wenn zu viel Flüssigkeit überläuft. Flowable kann helfen, dies nicht zu erreichen:

mit enormem Gegendruck:

Geben Sie hier die Bildbeschreibung ein

Bei Verwendung von fließfähigem Material ist der Gegendruck jedoch viel geringer:

Geben Sie hier die Bildbeschreibung ein

Rxjava2 verfügt über einige Gegendruckstrategien, die Sie je nach Anwendungsfall anwenden können. Mit Strategie meine ich, dass Rxjava2 eine Möglichkeit bietet, mit Objekten umzugehen, die aufgrund des Überlaufs (Gegendruck) nicht verarbeitet werden können.

Hier sind die Strategien. Ich werde nicht alle durchgehen, aber wenn Sie sich zum Beispiel keine Sorgen über die überfüllten Elemente machen möchten, können Sie eine Drop-Strategie wie die folgende verwenden:

Observable.toFlowable (BackpressureStrategy.DROP)

Soweit ich weiß, sollte es ein Limit von 128 Elementen in der Warteschlange geben, danach kann es zu einem Überlauf (Gegendruck) kommen. Auch wenn es nicht 128 ist, ist es nahe an dieser Zahl. Hoffe das hilft jemandem.

Wenn Sie die Puffergröße von 128 ändern müssen, sieht es so aus (aber beachten Sie alle Speicherbeschränkungen:

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

Bei der Softwareentwicklung bedeutet eine Gegendruckstrategie normalerweise, dass Sie den Emitter anweisen, etwas langsamer zu werden, da der Verbraucher die Geschwindigkeit Ihrer emittierenden Ereignisse nicht bewältigen kann.

j2emanue
quelle
Ich dachte immer, Gegendruck sei der Name für eine Familie von Mechanismen, mit denen der Verbraucher den Produzenten benachrichtigen könnte, um langsamer zu werden ...
kboom
Könnte der Fall sein. Ja
j2emanue
Gibt es Nachteile bei der Verwendung eines Flowable?
IgorGanapolsky
Diese Bilder lügen mich an. Das Löschen von Ereignissen führt nicht zu "mehr Geld" am unteren Rand.
EpicPandaForce
1
@ j2emanue, Sie verwechseln die Puffergröße für Operatoren und den Operator Flowable.buffer (int). Bitte lesen Sie die Javadocs sorgfältig durch und korrigieren Sie Ihre Antwort entsprechend: reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html
tomek
15

Die Tatsache, dass Sie Flowablenach der Ausgabe von 128 Werten ohne Gegendruckbehandlung abgestürzt sind, bedeutet nicht, dass es immer nach genau 128 Werten abstürzt: Manchmal stürzt es nach 10 ab, manchmal stürzt es überhaupt nicht ab. Ich glaube, dies ist passiert, als Sie das Beispiel ausprobiert haben Observable- es gab zufällig keinen Gegendruck, sodass Ihr Code normal funktionierte, beim nächsten Mal möglicherweise nicht. Der Unterschied in RxJava 2 besteht darin, dass es in Observables kein Konzept für Gegendruck mehr gibt und keine Möglichkeit, damit umzugehen . Wenn Sie eine reaktive Sequenz entwerfen, die wahrscheinlich eine explizite Behandlung des Gegendrucks erfordert, Flowableist dies die beste Wahl.

Egor
quelle
Ja, ich habe beobachtet, dass es manchmal nach weniger Werten brach, manchmal nicht. Aber wenn ich zum Beispiel nur intervalohne handhabe, backpressurewürde ich dann ein seltsames Verhalten oder Probleme erwarten?
user2141889
Wenn Sie sicher sind, dass in einer bestimmten beobachtbaren Sequenz keine Probleme mit dem Gegendruck auftreten können, ist es in Ordnung, den Gegendruck zu ignorieren.
Egor