如何使用最大并行请求数发送 1000 个 XHTTP 请求 [英] How to send 1000 XHTTP Requests with maximum number of parallel requests

问题描述

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

I have an Angular application, which needs to send N XHTTP requests, where 1 <= N <= 10000.

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

The application needs to handle it as fast as possible, so preferably there should be multiple active XHTTP requests at the same time, with a slide-window of multiple requests at the same time. Using WebSocket, or other streaming-like solution is not possible due to server-side API limitations.

我的第一个想法是使用类似 RxJS forkJoin，但我很难限制并发请求数.据我所知，最大请求数有 API 限制，例如 Chrome 将只允许 8 个同时请求.

My first idea was to use something like RxJS forkJoin, but I struggle to limit the concurrent requests number. As far as I know, there are API limitations for max requests number, for instance Chrome will allow only 8 simultaneous requests.

大多数解决方案/教程我发现 a.) 不限制最大并发连接数或 b.) 不动态更新(超时解决方案对于此任务效率不高).

Most of the solutions/tutorials I found either a.) does not limit the maximum number of concurrent connections or b.) does not update dynamically (timeout solutions are not efficient for this task).

例如:

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，如果一个连接完成，则立即启动新请求.

is not good for me, as the limitation is achieved by delay, but I would like to achieve something like a thread based solution: there are maximum of Y number of concurrent connections, and if one finishes, a new request starts immediately.

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 感觉应该有一个已经写好的解决方案.

Any ideas how to solve this nicely? RxJS feels like there should be a solution for this already written.

推荐答案

你可以尝试使用 RxJS bufferCount 和 concatMap 运算符以及 forkJoin().

You could try to use RxJS bufferCount and concatMap operators along with forkJoin().

来自 bufferCount 文档:

收集发出的值直到满足提供的数字，发出为数组.

Collect emitted values until provided number is fulfilled, emit as array.

因此它收集 n 个通知并将其作为数组发出.然后我们可以通过 forkJoin() 将数组传递给 n 个并行请求.

So it collects n number of notifications and emit it as an array. We could then pass the array through forkJoin() for n number of parallel requests.

试试下面的方法

我假设 this.urls 是类似于

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

然后请求触发代码看起来像

Then the requests triggering code would look like

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 的 this 评论工程师，最大的实际解决方案.连接到主机/域限制将使用 WebSockets 或域分片.但既然您提到在您的环境中这是不可能的，您可以使用缓冲请求解决方法.

---

According to this comment by a Chromium engineer, actual solution to the max. connections to host/domain limit would be to use WebSockets or Domain sharding. But since you mention it isn't possible in your environment, you could use the buffered request workaround.

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

However I wouldn't try to buffer to the max limit. If you were to send more requests to the same domain than the max allowed, you could see that the additional requests would actually throttle behind till the above requests are finished. So say if were to buffer to the max allowed limit, and your application sends an additional request to the same domain from somewhere else on which the app workflow depends on, the entire app could throttle.

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

So it's better to either use WebSockets or Domain sharding. And if neither is possible it's better to buffer the requests to a number of requests less than* the max allowed limit.

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

这篇关于如何使用最大并行请求数发送 1000 个 XHTTP 请求的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！