在 RxJava2 中将 Observable 转换为具有背压的 Flowable

Question

我正在观察NetworkResource 产生的行，将其包装在Observable.create 中。这是代码，为简单起见，缺少 try/catch 和取消：

fun linesOf(resource: NetworkResource): Observable<String> =
        Observable.create { emitter ->
            while (!emitter.isDisposed) {
                val line = resource.readLine()
                Log.i(TAG, "Emitting: $line")
                emitter.onNext(line)
            }
        }

问题是后来我想把它变成一个Flowable 使用observable.toFlowable(LATEST) 来增加背压，以防我的消费者跟不上，但根据我的做法，消费者会停止接收一个接一个的项目128.

A) 这样一切正常：

val resource = ...
linesOf(resource)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .toFlowable(BackpressureStrategy.LATEST)
    .subscribe { Log.i(TAG, "Consuming: $it") }

B) 这里消费者在 128 个项目后卡住（但发射仍在继续）：

val resource = ...
linesOf(resource)
    .toFlowable(BackpressureStrategy.LATEST)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe { Log.i(TAG, "Consuming: $it") } // <-- stops after 128

在选项 A) 中一切正常，没有任何问题，我可以看到 Emitting: ... 日志与 Consuming: ... 日志并排显示。

在选项 B) 中，我可以看到 Emitting: ... 日志消息愉快地发出新行，但我在第 128 项之后不再看到 Consuming: ... 日志消息，即使发出仍在继续。

问题：有人能帮我理解为什么会这样吗？

Answer 1

首先，您使用了错误的类型和错误的运算符。使用Flowable 消除了转换的需要。使用Flowable.generate 会给你带来背压：

Flowable.generate(emitter -> {
    String line = resource.readLine();
    if (line == null) {
        emitter.onComplete();
    } else {
        emitter.onNext(line);
    }
});

其次，您的版本挂起的原因是由于subscribeOn引起的同一个池死锁。来自下游的请求被安排在您的急切发射循环之后并且无法生效，在默认的 128 个元素处停止发射。使用Flowable.subscribeOn(scheduler, false) 来避免这种情况。