Rxjs: Observable.combineLatest против Observable.forkJoin

86

Просто интересно, в чем разница между Observable.combineLatestи Observable.forkJoin? Насколько я понимаю, единственная разница заключается в том, forkJoinчто Observables должны быть завершены, а combineLatestвозвращать последние значения.

rxjs rxjs5 Туонг Ле
источник
4
Примечание: В rxjs6 + это теперь только combineLatest()и forkJoin()функции , которые создают наблюдаемым. Они делают то же самое, но синтаксис отличается. Не путайте, combineLatestоткуда rxjs/operatorsидет «конвейерный» оператор. Если вы импортируете не тот, то получите ошибки.
Simon_Weaver

Ответы:

129

Он не только forkJoinтребует завершения всех входных наблюдаемых объектов, но также возвращает наблюдаемый объект, который производит одно значение, которое является массивом последних значений, созданных входными наблюдаемыми объектами. Другими словами, он ждет, пока завершится последний входной наблюдаемый объект, а затем выдает одно значение и завершается.

Напротив, combineLatestвозвращает Observable, который создает новое значение каждый раз, когда это делают входные наблюдаемые, после того как все входные наблюдаемые создали хотя бы одно значение. Это означает, что он может иметь бесконечное количество значений и может быть неполным. Это также означает, что входные наблюдаемые не должны завершаться перед созданием значения.

GregL
источник
@GregL есть ли функция, которая работает как forkJoin, но также будет работать с ошибочными http-вызовами?
Туккан
2
@Tukkan, я бы связал оператор, который восстанавливается после ошибок, с каждым наблюдаемым http, чтобы вы определяли значение (я) для использования для каждого запроса при ошибке, вместо того, чтобы искать оператор, который объединит несколько наблюдаемых, которые могут ошибаться (я не уверен, что такой оператор существует). Операторы, которые восстанавливаются после ошибок, включают .catch(), .onErrorResumeNext()и, возможно .retry()(если вызов Http может прерываться периодически).
GregL
1
Чтобы уточнить, оба они производят массивы.
Simon_Weaver
@Simon_Weaver Не обязательно. В случае combineLatest(), вы можете предоставить функцию проекции, чтобы указать, как создавать выходное значение из последних значений из входных наблюдаемых. По умолчанию, если вы его не укажете, вы получите массив последних сгенерированных значений.
GregL
Каждое излучение от combLatest представляет собой массив собранных значений. Я просто хотел добавить это, поскольку вы упомянули только массив для forkjoin. Таким образом, они такие же. И да, с RxJS всегда есть тонкости, поэтому, если я что-то пропустил, вы можете пояснить, что вы имели в виду.
Simon_Weaver
14

Также:

combLatest (...) последовательно запускает наблюдаемые объекты, один за другим

combineLatestвнутреннее использование concat, что означает, что значение должно быть получено для каждого наблюдаемого в массиве, прежде чем переходить к следующему.

// partial source of static combineLatest (uses the rxjs/operators combineLatest internally):
// If you're using typescript then the output array will be strongly typed based on type inference
return function (source) { return source.lift.call(from_1.from([source].concat(observables)), 
                           new combineLatest_1.CombineLatestOperator(project)); };

forkJoin (...) запускает наблюдаемые параллельно, в то же время

Если у вас есть 3 исходных наблюдаемых объекта, и для каждого из них требуется 5 секунд для запуска, то для запуска потребуется 15 секунд combineLatest. В то время как forkJoinони выполняются параллельно, это займет 5 секунд.

Так forkJoinработает больше похоже на то, Promise.all(...)где порядок не соблюдается.

Рекомендации по обработке ошибок:

Если какая-либо из наблюдаемых ошибок вышла из строя - combineLatestпоследующие не будут выполнены, но с forkJoinними все работают. Таким образом, combineLatestможет быть полезно выполнить «последовательность» наблюдаемых и собрать все результаты вместе.

Дополнительное примечание: если исходные наблюдаемые объекты уже «запущены» (на которые подписано что-то еще) и вы используете shareих, то вы не увидите такого поведения.

Еще более сложное примечание: CombineLatest всегда будет предоставлять вам последнюю информацию из каждого источника, поэтому, если один из наблюдаемых источников генерирует несколько значений, вы получите последнее. Он не просто получает одно значение для каждого источника и переходит к следующему. Если вам нужно убедиться, что вы получаете только «следующий доступный элемент» для каждого наблюдаемого источника, вы можете добавить его .pipe(take(1))к наблюдаемому источнику по мере добавления его во входной массив.

Simon_Weaver
источник
Спасибо за рассмотрение ошибок. Обработку ошибок довольно сложно понять в случае нескольких наблюдаемых.
D Deshmane 01
2
Я считаю , что вы неправильно поняли , что concat()в combineLatest()коде делают. Мне кажется, что это Array.prototype.concatметод, а не concatметод RxJS . Если предположить, что я прав, то этот ответ вводит в заблуждение и неверен, так как combineLatest()не выполняет наблюдаемые последовательно один за другим. Он выполняет их параллельно, так же, как forkJoin(). Разница в том, сколько значений создается и должны ли исходные наблюдаемые завершаться или нет.
GregL 05
@GregL Array.concat бессмысленен с наблюдаемыми. Я полностью полагаюсь на последовательный характер concat - главное, что нужно понять, это то, что вы не получите никаких выходных значений, пока все они не будут выполнены.
Simon_Weaver 05
Я имел в виду этот код в опубликованном вами образце исходного кода:, [source].concat(observables)предполагая, что это именно то, что вы имели в виду, говоря, что « combineLatestвнутреннее использование concat». Мне кажется, вы путаете объединение массивов с объединением RxJS. Последний действительно последовательно выполняет входные наблюдаемые, ожидая завершения каждого из них. Но это не используется combineLatest(), это отдельный оператор целиком.
GregL 05
3
Я считаю, combineLatest()и forkJoin()оба работают параллельно. Выполнение приведенного ниже кода дает около 5000 для обоих. const start = new Date().getTime(); combineLatest([of(null).pipe(delay(5000)), of(null).pipe(delay(5000)), of(null).pipe(delay(5000))]).subscribe(() => console.log(new Date().getTime() - start)); forkJoin([of(null).pipe(delay(5000)), of(null).pipe(delay(5000)), of(null).pipe(delay(5000))]).subscribe(() => console.log(new Date().getTime() - start));
Джереми
13

forkJoin - Когда все наблюдаемые завершены, испускает последнее переданное значение из каждого.

combLatest - когда какой-либо наблюдаемый излучает значение, испускает последнее значение из каждого.

Использование очень похоже, но вы не должны забывать отказаться от подписки на combLatest, в отличие от forkJoin .

Дмитрий Гринько
источник
1
если один, если запрос не выполняется, все вложенные запросы автоматически отменяются, как я могу решить эту проблему?
Сунил
1
@SunilGarg Вы должны использовать forkJoin илиcommonLatest, если хотите получить только все результаты. Если вас не интересуют все результаты, вам следует использовать подписки отдельно.
Дмитрий Гринько
Сунил
Можете ли вы объяснить, почему требуется отказ от подписки, на примере кода
Сунил Гарг,