RxJS 相当于 Async.js mapLimit

Question

Async.js mapLimit 及其系列 <name>Limit 函数基本上像信号量一样工作：它们允许有限数量的任务同时运行而额外的传入任务被添加到队列中。队列成为（冷？连接？）生产者。任务运行ner 在有可用位置时立即从队列中排出一个项目（其中一个任务完成）。

这样有限数量的并发任务始终处于活动状态。

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

RxJS 相当于 Async.js mapLimit

RxJS equivalent of Async.js mapLimit

javascript

asynchronous

rxjs

async.js