可观察与可流动rxJava2

Question

我一直在寻找新的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我在管理和处理消费方面有哪些优势？

Answer 1

实际上，反压显示的是有界缓冲区，Flowable.observeOn缓冲区中包含128个元素，其流失速度与dowstream可以吸收的速度一样快。您可以单独增加此缓冲区的大小以处理突发信号源，并且所有反压管理实践仍从1.x开始适用。Observable.observeOn具有无限制的缓冲区，该缓冲区会不断收集元素，并且您的应用程序可能会耗尽内存。

您可以使用Observable例如：

• 处理GUI事件
• 使用短序列（总共少于1000个元素）

您可以使用Flowable例如：

• 冷源和非定时源
• 发电机一样的来源
• 网络和数据库访问器