如何以最大并行请求数发送 1000 个 XHTTP 请求

Question

我有一个 Angular 应用程序，需要发送 N 个 XHTTP 请求，其中 1 <= N <= 10000。

应用程序需要尽可能快地处理它，因此最好同时有多个活动的 XHTTP 请求，并具有同时多个请求的滑动窗口。由于服务器端 API 限制，无法使用 WebSocket 或其他类似流式处理的解决方案。

我的第一个想法是使用RxJS forkJoin之类的东西，但我很难限制并发请求数。据我所知，API 对最大请求数有限制，例如Chrome 只允许 8 个并发请求。

我发现的大多数解决方案/教程要么 a.) 不限制最大并发连接数，要么 b.) 不动态更新（超时解决方案对此任务效率不高）。

例如：

const test = () =>
  request(`https://swapi.co/api/people/1/`)
    .pipe(
      delay(1000),
      switchMap(response => from(response.films)),
      concatMap((url: string) => request(url).pipe(delay(1000))),
      scan((acc, res) => [...acc, res.title], []),
      tap(console.log)
    )
    .subscribe()

这对我来说不好，因为限制是通过延迟实现的，但我想实现类似基于线程的解决方案：最多有 Y 个并发连接，如果一个连接完成，则立即启动一个新请求。

const test = () =>
  request(`https://swapi.co/api/people/1/`)
    .pipe{
      switchMap(response => from(response.films)),
      specialOperatorIAmLookingFor((url: string) => request(url), 8),   // where '8' is the maximum number of paralell requests
      scan((acc, res) => [...acc, res.title], []),
      tap(console.log)
    )
    .subscribe()

有什么想法如何很好地解决这个问题吗？RxJS 感觉应该有一个已经写好的解决方案。

Answer 1

您可以尝试将 RxJSbufferCount和concatMap运算符与forkJoin().

来自bufferCount文档：

收集发出的值，直到满足提供的数量为止，以数组形式发出。

因此它收集n大量通知并将其作为数组发出。然后我们可以通过该数组来获取并行请求的forkJoin()数量n。

尝试以下操作

我假设this.urls是类似于 HTTP 请求的集合

urls = [
  this.http.get('url1'),
  this.http.get('url2'),
  this.http.get('url3'),
  ...
];

那么请求触发代码将如下所示

bufferedRequests() {
  from(this.urls).pipe(
    bufferCount(6),      // <-- adjust number of parallel requests here
    concatMap(buffer => forkJoin(buffer))
  ).subscribe(
    res => console.log(res),
    err => console.log(err),
    () => console.log('complete')
  );
}

根据Chromium 工程师的评论，实际解决方案最大。与主机/域的连接限制将使用 WebSockets 或域分片。但既然您提到这在您的环境中是不可能的，您可以使用缓冲请求解决方法。

但是我不会尝试缓冲到最大限制。如果您向同一域发送的请求多于允许的最大请求数，您会发现额外的请求实际上会受到限制，直到上述请求完成为止。因此，如果要缓冲到允许的最大限制，并且您的应用程序从应用程序工作流程所依赖的其他地方向同一域发送额外的请求，则整个应用程序可能会受到限制。

因此，最好使用 WebSockets 或域分片。如果两者都不可能，最好将请求缓冲到小于*最大允许限制的请求数量。

^{* 显然，如果您 100% 确定在缓冲过程中不会向同一域触发其他请求，那么您可以缓冲到最大。允许的限度。}