使用RxJS進行內存管理Observable.concatMap？

Question

我是RxJS的新手，還在學習如何使用庫。爲concatMap文檔提供了以下警告：

警告：如果源值到達不休，比他們的 相應的內觀測量速度可以完成的，它會導致內存 問題，如內觀測量在無界緩衝區聚斂等待 他們將被訂閱。

這對我來說是一個問題，因爲我有一個內存密集但快速的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中處理這類問題的正確方法是什麼？

Answer 1

這是一個典型的生產者 - 消費者問題。 您可以使用背壓操作員來限制發送進行處理的元素的數量。請參閱controlled streams。