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

Question

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

我正在使用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/flatMap 时设置最大并发请求，即真不错。

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

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

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

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

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

Answer 1

.flatMap()，又名.mergeMap() 有一个可选参数“并发”

mergeMap(project: function(value: T, ?index: number): Observable, resultSelector: function(outerValue: T, innerValue: I, outerIndex: number, innerIndex: number): any, concurrent: number )

所以这只是一个问题：

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 个并发请求