使用 RxJS 和 Angular2 限制并发 http 请求

我以前见过这个问题，但回答总是让我不满意，所以我会尽量准确：

我在用着https://www.npmjs.com/package/rxjs https://www.npmjs.com/package/rxjs想要对大量的 http 请求进行排队，并且我想将它们链接到其他操作中，如下所示：

urls$.flatMap(x => fetchUrls(x)).subscribe(x => console.log(x));

url$ 是一个 Observable，而 fetchUrls 是一个返回新 Observable 的方法。这会工作得很好，除了来自 urls$ 的第一个流非常快并且 fetchUrls 方法很慢（执行 http 请求/响应）。我最终同时运行了 200 多个 http 请求。

我认为在客户端上留下 200 多个待处理的 http 请求会带来麻烦，所以我想以某种方式集中这些请求，并且我读到可以在使用时设置最大并发请求mergeMap http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html#instance-method-mergeMap/flatMap，这真的很好。

我在这里为这个东西创建了一个游乐场：http://www.webpackbin.com/EkQXtBEwz http://www.webpackbin.com/EkQXtBEwz我有一个名为 fakeHttpLookup 的方法，它将响应延迟一秒，并为其提供 200 个 url 列表 - 如果按顺序运行将需要 200 秒。我猜想，当 maxConcurrent 设置为 5 时，最好的情况是它会在 40 秒内处理整个流。

问题是 - 我无法让它工作。它确实需要前 5 个，但随后停止。它永远不会完成整个流。

我希望看到计数器每经过一秒就会增加 5，说明一次有 5 个并发 http 请求。

有人可以帮忙吗？我可以忍受一次只运行一个请求 - 我可以一次缓冲 5 个请求，但这里重要的是不要一次排队所有 200 个请求 - 或者只获取前几个请求。

总而言之： 我希望能够对 http 请求进行排队/池化，并且流必须能够限制请求数量，而无需猜测硬编码的等待延迟。

.flatMap(), aka .mergeMap() http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html#instance-method-mergeMap有一个可选参数“并发”

mergeMap（项目：函数（值：T，？索引：数字）：Observable，resultSelector：函数（outerValue：T，innerValue：I，outerIndex：数字，innerIndex：数字）：任何，并发：数量)

所以这只是一个问题：

function fetch(id) {
  return Rx.Observable.of(id)
    .do(i => console.log(`fetching request ${i}`))
    .delay(2 * 1000);
}

Rx.Observable.range(1, 20)
  .flatMap(id => fetch(id), null, 5)
  .subscribe(console.log);

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.3/Rx.js"></script>

执行 5 个并发请求