带有承诺的 Rx.js 并发答案

【问题标题】：Rx.js concurrency with promises带有承诺的 Rx.js 并发
【发布时间】：2015-11-21 15:01:33
【问题描述】：

我想通过一系列异步/网络操作（远程 HTTP 请求）移动对象数组来处理它们。

在其中一些操作中，我希望确保同时处理的项目不超过 X 个。

我怎样才能做到这一点？

示例代码：

function someAsyncOp(item) {...} // returns a promise

var source = Rx.Observable.from([{item1},{item2},...])
source
  .flatMap((item) => {

    // I WANT THE FOLLOWING OPERATION TO BE EXECUTING  
    // ON AT MAX 10 ITEMS AT A TIME, NEXT ITEM SHOULD
    // BE SUBMITTED ONLY WHEN A SLOT GETS FREED AS A 
    // RESULT OF THE PROMISE SUCCEEDING OR FAILING

    return Rx.Observable.fromPromise(someAsyncOp(item))

  })
  .subscribe(
    console.log, 
    console.error, 
    () => console.log('completed')
  )

【问题讨论】：

把take(X)放在.flatMap之前see interactive diagramm
@valery.sntx 这将在第一个 X 之后停止我的源流，这意味着永远不会处理所有后续项目。我猜。我希望处理所有项目。

标签： javascript concurrency promise rxjs

【解决方案1】：

flatMap 有一个名为flatMapWithMaxConcurrent 的兄弟，它接受一个并发参数。它在功能上类似于 Benjamin 的回答所建议的 map(fn).merge(n)。

function someAsyncOp(item) {...} // returns a promise

var source = Rx.Observable.from([{item1},{item2},...])
source
   //Only allow a max of 10 items to be subscribed to at once
  .flatMapWithMaxConcurrent(10, (item) => {

    //Since a promise is eager you need to defer execution of the function
    //that produces it until subscription. Defer will implicitly accept a promise
    return Rx.Observable.defer(() => someAsyncOp(item))

    //If you want the whole thing to continue regardless of exceptions you should also
    //catch errors from the individual processes
                        .catch(Rx.Observable.empty())
  })
  .subscribe(
    console.log, 
    console.error, 
    () => console.log('completed')
  )

【讨论】：

我认为.defer(... 部分是我正在寻找的。错误处理部分也很有用
在withLatestFrom 代替flatMap 的情况下如何管理并发有什么想法吗？

【解决方案2】：

您可以将merge 与map 一起使用，而不是flatMap：

var concurrency = 10;
source.map(someAsyncOp).merge(concurrency).subscribe(x => console.log(x));

请注意，由于 promises 是急切的，而 observables 是惰性的，因此 fromPromise 不会削减它（并且 Rx 无论如何都可以在没有它的情况下同化 promises）。我建议将其包装在 create 中。

var delay = function(ms){ return new Promise(function(r){ setTimeout(r, 2000, ms) }); }

var log = function(msg){ document.body.innerHTML += msg + "<br />"; }

Rx.Observable.range(1000, 10).map(delay).merge(2).subscribe(log)

&lt;script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/4.0.7/rx.all.js"&gt;&lt;/script&gt;

【讨论】：