javascript - RxJS 相当于 Async.js mapLimit

Question

Async.js mapLimit及其家族 <name>Limit函数基本上像信号量一样工作:它们允许有限数量的任务同时运行，同时将额外的传入任务添加到队列中。队列成为(冷？连接？)生产者。任务运行器会在有空位(其中一个任务完成)后立即从队列中排出一个项目。

这样，有限数量的并发任务始终处于事件状态。

如何在 RxJS 中实现类似的功能？

Answer 1

defer 和 flatMapWithMaxConcurrent 的组合是 RxJs 的实现方式:

// returns a promise
function runSomeJob(input) { ... }

function runSomeJobObservable(input) {
    return Rx.Observable.defer(function () {
        return runSomeJob(input);
    });
}

var inputStream = // some Rx.Observable

// only allow 5 jobs to run concurrently
var outputStream = inputStream
    .flatMapWithMaxConcurrent(5, runSomeJobObservable);

ouputStream.subscribe(...);