如何一次处理RxJS流n项目，一旦项目完成后，又自动填充回n？

Question

我有一个事件流，我想调用一个函数来返回每个事件的承诺，问题是这个函数非常昂贵，所以我想一次最多处理n个事件。

这卵石图可能是错的，但是这是我想什么：

---x--x--xxxxxxx-------------x-------------> //Events 
---p--p--pppp------p-p-p-----p-------------> //In Progress 
-------d--d--------d-d-dd------dddd--------> //Promise Done 

---1--21-2-34-----------3----4-3210-------- //QUEUE SIZE CHANGES 

这是代码，我到目前为止有：

var n = 4; 
var inProgressCount = 0; 

var events$ = Rx.Observable.fromEvent(produceEvent, 'click') 
    .map((ev) => new Date().getTime()); 

var inProgress$ = events$.controlled(); 

var done$ = inProgress$  
    .tap(() => inProgressCount++) 
    .flatMap((timestamp) => Rx.Observable.fromPromise(expensiveComputation(getRandomInt(1, 5) * 1000, timestamp))); 

done$.subscribeOnNext((timestamp) => { 
    inProgressCount--; 
    inProgress$.request(Math.max(1, n - inProgressCount)); 
}); 

inProgress$.request(n); 

有此代码的两个问题：

1. 它使用了inProgressCount变量，它是用边 更新效果函数秒。
2. 完成$订阅仅在请求超过1个受控流的项目时调用一次。这使得inProgressCount var更新不正确，这最终将队列限制为一次一个。

你可以看到它在这里工作： http://jsbin.com/wivehonifi/1/edit?js,console,output

问题：

1. 有没有更好的方法吗？
2. 我该如何摆脱inProgressCount变量？
3. 为什么完成$订阅只需要在请求多个项目时调用一次？

更新：
回答问题＃3：switchMap是一样的flatMapLatest，所以这就是为什么我只获得了最后一个。将代码更新为flatMap而不是switchMap。

Answer 1

实际上根本不需要使用背压。有一个名为flatMapWithMaxConcurrent的运营商为您提供此服务。它本质上是调用.map().merge(concurrency)的别名，它一次只允许最大数量的流在运行。

我更新了jsbin这里：http://jsbin.com/weheyuceke/1/edit?js,output

但我注释以下重要的一点：

const concurrency = 4; 

var done$ = events$ 
    //Only allows a maximum number of items to be subscribed to at a time 
    .flatMapWithMaxConcurrent(concurrency, 
    ({timestamp}) => 
     //This overload of `fromPromise` defers the execution of the lambda 
     //until subscription      
     Rx.Observable.fromPromise(() => { 
     //Notify the ui that this task is in progress         
     updatePanelAppend(inProgress, timestamp); 
     removeFromPanel(pending, timestamp); 
     //return the task 
     return expensiveComputation(getRandomInt(1, 5) * 1000, timestamp) 
    }));