如何一次处理 RxJS 流 n 个项目，并在完成一个项目后，再次自动填充回 n 个？

Question

我有一个事件流，我想调用一个函数，该函数 returns 对每个事件的承诺，问题是这个函数非常昂贵，所以我想在一次最多 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);

这段代码有两个问题：

它使用的是 inProgressCount var，它用 side 更新效果函数。
~~当我从受控流中请求超过 1 个项目时，仅调用完成 $ 订阅一次。这使得 inProgressCount var 更新不正确，这最终将队列一次限制为一个。~~

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

问题：

有没有更好的方法？
如何删除 inProgressCount 变量？
~~为什么 done$ 订阅在请求多个项目时只被调用一次？~~

更新：
问题 #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)
     }));

如何一次处理 RxJS 流 n 个项目，并在完成一个项目后，再次自动填充回 n 个？

How to process RxJS stream n items at a time and once an item is done, autofill back to n again?

javascript

rxjs

backpressure