Будут ли параллельные запросы с RXJS улучшать производительность?

Question

Проблема: мне нужно скачать огромный файл (5-10 ГБ) с простыми XHR-запросами, чтобы упростить вопрос, мне нужно сделать как минимум: 100 запросов (может быть больше). Я хотел бы улучшить производительность с помощью RXJS (параллельные http-запросы).

Вопросы:

1. Можно ли вообще улучшить производительность с помощью Parallel?
2. Разве распараллеливание запросов не будет таким же, как использование 1 запроса для каждой итерации цикла запроса? Пропускная способность одинакова - НЕТ?
3. Есть ли другие способы улучшить мой перф?

Stack:

• Угловой 7,2.х
• Машинопись 3.2.4
• Rxjs 6.4.0

Чего я на самом деле достиг: Что мне действительно нужно, так это применить синхронизацию к mergeMap, чтобы включить или выключить ее, например, у меня есть 100urls, но я не хочу, чтобы они запускались до последнего из них.

Например, я хочу, чтобы он работал до 17-го числа, но приостановит выполнение запросов и будет ждать завершения определенной логики и запроса.

Пример кода:

from(observableQueries)
           .pipe(
               mergeMap(partialQuery => this.messageService.getResults(partialQuery, true), 4),
               takeWhile(o => {
                   if (!o.isMoreResults && o.auditMessageList.length === 0) {
                       this.logger.log(`First page that returns false ${ o.number }`);
                       this.logger.log(`Count of responses that exists in array: ${ allResults.length }`);
                       if (!firstPageThatShouldBeStopped) {
                           firstPageThatShouldBeStopped = o.number;
                       }

                       if (allResults.length === firstPageThatShouldBeStopped) {
                           return false;
                       }
                   }
                   return true;
               }),
               retryWhen(genericRetryStrategy()),
               catchError((err, caught) => {
                   this._alertService.error(this._translate.instant('EXPORT_TO_CSV_DOWNLOAD_ERROR'));
                   return throwError(err);
               })
           )
           .subscribe(
               (res: MessagesResult) => {
                   if (reThrowError) {
                       this.logger.info('Error will be thrown for e2e purposes');
                       throw new Error('Error will throw for e2e purposes');
                   }
                   if (res.isMoreResults && res.auditMessageList.length > 0) {
                       allResults[res.number] = res;
                       this.subject.next(true);
                   } else if (!res.isMoreResults && res.auditMessageList.length > 0) {
                       allResults[res.number] = res;
                       this.subject.next(true);
                   } else {
                       this.subject.next(false);
                   }
               },
               err => {
                   // clear subject after emitting value...
                   this.subject.next(true);
                   return this.handleError(err);
               },
               () => {
                   this.logger.info('Download file finished...');
                   this.logger.info('Time consumed: ', performance.now() - start);
                   try {
                       this.logger.info(`Count Responses: ${ allResults.length } `);
                       const allResultSorted = this._sortResults(allResults);
                       let counter = 0;
                       for (let i = 0; i < allResultSorted.length; i++) {
                           this.logger.info('Index: ' + i, allResultSorted[i]);
                           counter += allResultSorted[i].auditMessageList.length;
                           this.logger.info('Length OF Messages: ' + i, counter);
                           this.fileSaver.save(!allResultSorted[i].isMoreResults, allResultSorted[i].auditMessageList,
                               `audit-events_${ LOCAL_QUERY_COPY.application }_${ timestamp }_${ moment()
                                   .tz(this._timezoneService.timezone).zoneName() }.csv`, null, headers);
                       }
                       this.subject.next(false);
                   } catch (e) {
                       this._alertService.error(this._translate.instant('EXPORT_TO_CSV_DOWNLOAD_ERROR'));
                       return this.handleError(e);
                   } finally {
                       // clear subject after emitting value...
                       this.subject.next(true);
                   }
               }
           );

Код работает! Но проблема в избыточных вызовах. Как можно заставить их остановиться, пока не будут выполнены последние запросы.

Answer 1

1. Да, конечно, только если ваш сервер способен обрабатывать желаемый параллелизм

2. 1 запрос для каждой итерации цикла запроса является последовательным запросом после одногои другие, так что это не то же самое, что распараллеливание

3. Обычно, если у вас 100 запросов, мы ограничим параллелизм, чтобы сервер номеров не зависал при чрезмерных запросах, продемонстрируйте код ниже

     const urls=[url1,url2,url3]
     const concurrency=5
     const fetchUrls=urls.map(url=>defer(_=>fetch(url)))
     merge(...fetchUrls,concurrency).subscribe(console.log)

Answer 2

Короткий ответ: да, параллелизм должен улучшить положение, до некоторой степени.

Быстрый поиск Google "пределов запросов браузера" поднимает после , что показывает, что вы можетеделать от 2 до 13 одновременных запросов на домен в зависимости от браузера.Пока интернет-соединение пользователя не насыщено, и вы еще не достигли этого предела, вы действительно должны иметь возможность делать и ожидать запросы одновременно.Это будет быстрее, чем отправлять запросы по одному, поскольку браузеру на самом деле не придется выполнять какую-либо работу, пока он ждет следующего.

Пропускная способность действительно такая же, но пропускная способность выше.

То, что улучшит вашу производительность, зависит от вашей конкретной ситуации, но я предполагаю, что выполнение 100 маленьких запросов на самом деле медленнее, чем одного большого, так как каждый маленький запрос должен выполнить отдельное рукопожатие, прежде чем он может начатьсяполучение данных.

Что касается RxJ, мне нравится ответ Fan Cheung , но я вижу, что у вас есть механизм повторных попыток, которого нет у их ответа.Должно быть достаточно легко добавить повтор для каждого из вызовов «извлечения» или сохранить список неудачных URL-адресов и повторить попытку после первой попытки пакета.Опять же, это зависит от ваших конкретных обстоятельств.

Что именно вы подразумеваете под избыточными вызовами?Какое поведение вы хотели бы видеть, что вы не видите?

Answer 3

Вы можете использовать RXJS оператор слияния: Предполагая, что getData (url) является методом, делающим запрос, и что этот метод возвращает Observable, вы можете сделать:

const urls: string[] = {url1, url2, url3,...};
let mergeHttpCallObservalbe: Observable<any>;
urls.forEach((url: string) => {
  const newHttpCallObservalbe : Observable<any> = myService.getData(url);
  if (mergeHttpCallObservalbe){
    mergeHttpCallObservalbe = merge(mergeHttpCallObservalbe, newHttpCallObservalbe);
  } else {
    mergeHttpCallObservalbe = newHttpCallObservalbe;
  }
});

// Now you have merged all your Observable, you can subscribe:
mergeHttpCallObservalbe.subscribe(result => {
  // Do your stuff...
});

Вот хорошая статья об операторах Rxjs: https://blog.angularindepth.com/learn-to-combine-rxjs-sequences-with-super-intuitive-interactive-diagrams-20fce8e6511

Надеюсь, это поможет

Answer 4

Разве распараллеливание запросов не будет таким же, как использование 1 запроса для каждой итерации цикла запроса?Пропускная способность одинакова - НЕТ

=> Я бы сказал НЕТ, потому что на сервере, когда один поток выполняет первый запрос, второй протектор может загрузить второй запрос => Для меня это лучшераспараллелить, но может я ошибаюсь