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(...);