使用flatMap和多个observeOn的链条时，背压没有得到尊重

Question

我不知道为什么背压在这个流动链中没有得到尊重。

我将其简化为以下三个阶段：heavyFlowableBasedOnInput，singleThreadedPieceOfWork和writeFileToDisk。前两个阶段的速度比最后一个阶段快得多，因此它们的输出会让内存等待第3步，从而导致内存不足错误/内存使用量不断增加。

当我观看调试器时，我注意到步骤1比步骤3更频繁地发生，尽管我已经将缓冲区大小（observeOn的第三个参数）设置为一个较低的数字（例如1）。

Flowable.fromIterable(listOfResources)
        .subscribeOn(Schedulers.io())
        .flatMap { flowableUsing(it) } // small flowable though, ~0-10, mostly 1 to 2 emissions
        .observeOn(Schedulers.single(), false, 1)
        .map { singleThreadedPieceOfWork() }
        .observeOn(Schedulers.io(), false, 1)
        .map { writeFileToDisk() }
        .subscribe()

我已经阅读了我正在使用的每个运算符的留档，看起来他们都有一些背压支持（例如通过背压BackpressureKind.PASS_THROUGH，荣誉下游背压BackpressureKind. FULL）。我没有线索说明为什么我要重载写入文件磁盘的队列/缓冲区。

我确实读过为什么我的RxJava Flowable在使用观察时不尊重背压？我知道观察会创建一个“异步边界”，但我不知道这如何影响背压。

Answer 1

来自akarnokds博客：

背压是RxJava的一个基石，它通过反应流防止缓冲区膨胀。大多数没有计时功能的运营商都会应用并尊重背压。不幸的是，flatMap在默认情况下只能说它尊重背压，但不能将其应用于主要输入。

这意味着，当您使用flatMap或merge的公共重载时，操作符将请求Long。从上游获取最大价值，同时实现所有价值。这种无界行为会导致对生成的内部观察值的活动订阅数量无界。

如果内部可观察对象很短或不经常发射，此属性不会引起太多麻烦，但是如果在平面图之后有一个异步边界，比方说，观察，项目很容易在平面图中堆积，并大大降低性能。

我想我通过在flatMap重载函数中指定maxConcurrency参数：kotlin解决了这个问题

Flowable.fromIterable(listOfResources)
        .subscribeOn(Schedulers.io())
        .flatMap({ flowableUsing(it) }, false, 1, 5) // <-- the maxConcurrency and bufferSize 
        .observeOn(Schedulers.single(), false, 1)
        .map { singleThreadedPieceOfWork() }
        .observeOn(Schedulers.io(), false, 1)
        .map { writeFileToDisk() }
        .subscribe()

或者，这不是一个解决方案。我可能只是降低了性能，以至于不再出现内存问题。

等待更多阅读/理解