使用 RxJS Observable.concatMap 进行内存管理？答案

【问题标题】：Memory management with RxJS Observable.concatMap?使用 RxJS Observable.concatMap 进行内存管理？
【发布时间】：2016-12-16 20:40:20
【问题描述】：

我是 RxJS 的新手，所以还在学习如何使用这个库。 concatMap 的文档给出了以下警告：

警告：如果源值无休止地到达并且比它们的值更快对应的内部Observables可以完成，会导致内存内部 Observables 堆积在无限缓冲区中等待的问题轮到他们订阅了。

这对我来说是个问题，因为我有一个内存密集型但快速的 concatMap 提供了一个缓慢的 concatMap。它是这样设置的：

let uploadObs = Observable.range(0, blockCount).concatMap(blockIndex => {
    // This part is fast and memory intensive. I'd like to use
    // a bounded buffer here or something similar to control
    // memory utilization

    let blockReaderObs = ...;
    // ... read a block from a large file object in blockReaderObs
    return blockReaderObs;
}).concatMap((blockData, index) => {
    // This part involves a POST so is much slower than reading a
    // file block
    let objFromBlockData = someTransformation(blockData);
    return this.http.post(someUrl, objFromBlockData)
        .map(transformResponse);
});

在 RxJS 中处理此类问题的正确方法是什么？

【问题讨论】：

标签： rxjs rxjs5

【解决方案1】：

这是一个经典的生产者-消费者问题。您可以使用背压操作符来限制要处理的元素数量。见controlled streams。

【讨论】：

controlled 仅存在于 RxJS 4
没有太多关于 RxJS5 中背压的文档

【解决方案2】：

我在使用 concatMap 和 concatAll 时遇到了类似的问题。我意识到我正在创建太多等待订阅的可观察对象。

在单独的 SO 讨论中的这个答案有助于编写我自己的解决方案来限制我在 concatMap 中的投入：

https://stackoverflow.com/a/40845089/181961

【讨论】：