Angular2 RxJS ForkJoin 替代方案答案

【问题标题】：Angular2 RxJS ForkJoin AlternativeAngular2 RxJS ForkJoin 替代方案
【发布时间】：2018-03-06 08:28:50
【问题描述】：

使用 forkJoin 时，所有项目（在本例中为 HTTP 请求）都会立即发送到服务器，然后作为数组发出：

let observable_array = [observable1, observable2, ...];

Observable.forkJoin(observable_array).subscribe(res => {
  console.log('Res: ', res);
}, err => {
  console.log('Err: ', err);
});

这对我来说是个问题，因为我使用的是单线程服务器 (NodeJS)。如果observable_array 包含 10 个 HTTP 请求，每个请求需要 3 秒才能完成，则 30 秒内不能发出其他请求。

我正在寻找的是一种执行请求1的方法，等待来自服务器的响应完成，然后执行请求2，然后等待响应等。使用这种方法意味着新请求不会必须等待 30 秒才能完成 observable_array 才能被服务器执行...它们最多只会等待 3 秒。

我想出了以下解决方案，但我相信知识渊博的人会有更优雅的解决方案。

let observable_array = [observable1, observable2, ...];

var queue = function(t, ob) { // Needed to use var instead of let because of scoping...

  if (ob) {

    ob.subscribe(res => {
      console.log('Res: ', res);
      queue(t, observables.shift());
    }, err => {
      console.log('Err: ', err);
    });

  }

}

queue(this, observables.shift());

【问题讨论】：

Sequential subscription to an array of observables的可能重复
我不确定他的问题是否相同 - 链接的“重复”不关心并发执行，而 OP 确实关心 - 他只关心按顺序订阅结果。不同的是关心太多并发的http请求...
你应该在你的后端寻找一个解决方案，因为如果有人想溢出你的调用堆栈，这很容易......
只需使用concat 而不是forkJoin

标签： node.js angular rxjs

【解决方案1】：

您可以使用自己喜欢的 flatMap/mergeMap 风格并添加并发参数。这将限制您正在服务的并发 observables/requests 的数量。

//emit value every 1s
const source = Rx.Observable.interval(1000);

const example = source.mergeMap(
    //project
    val => Rx.Observable.interval(5000).take(2),
  //resultSelector
  (oVal, iVal, oIndex, iIndex) => [oIndex, oVal, iIndex, iVal],
  //concurrent
  2 
);
/*
        Output:
        [0, 0, 0, 0] <--1st inner observable
        [1, 1, 0, 0] <--2nd inner observable
        [0, 0, 1, 1] <--1st inner observable
        [1, 1, 1, 1] <--2nd inner observable
        [2, 2, 0, 0] <--3rd inner observable
        [3, 3, 0, 0] <--4th inner observable
*/
const subscribe = example.subscribe(val => console.log(val));

取自https://www.learnrxjs.io/operators/transformation/mergemap.html

注意 2 作为最后一个参数，以将并发请求限制为最多两个。显然，您可以将其更改为 1。上面的示例使用 .interval 模拟 observables - 您只需将其替换为您自己的 observable 列表。

【讨论】：