Создает ли конвейер оператора промежуточные наблюдаемые, аналогичные фильтрующим, отображающим, уменьшающим, создают промежуточные массивы?

Question

Мне нужны разъяснения об эффективности конвейеров операторов в Rx JS.

Насколько мне известно о конвейере операторов в Rx JS, каждый оператор внутри конвейера получает наблюдаемый и создает новую (возможно модифицированную) наблюдаемую, которая возвращается в качестве следующего оператора. Такое поведение было бы похоже на JavaScript фильтр, отображение, уменьшить поведение. Таким образом, оставляя исходный наблюдаемый (поток) или массив нетронутым / чистым.

Это предположение поддерживается Документацией Rx JS по адресу: https://rxjs-dev.firebaseapp.com/guide/operators

Pipeable Operator - это функция, которая принимает Observable в качестве входных данных и возвращает другое Observable. Это чистая операция: предыдущая наблюдаемая остается неизменной.

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

Кроме того, я читаю книгу «Реактивное программирование с использованием Rx JS 5», написанное Sergi Mansilla. Я знаю, что Rx JS в настоящее время имеет версию 6.5.3, но я ожидаю, что механизм basi c с тех пор не изменился.

В книге есть раздел об эффективности конвейера, в котором говорится что наблюдаемые трубопроводы не создают промежуточных наблюдаемых. Вместо этого они применяют все операции к каждому элементу в одном go. Это имеет смысл, поскольку оператор take (amount) завершает наблюдаемый поток после получения первых amount элементов. Это также объясняет ленивую оценку , пересекающую исходный наблюдаемый поток только один раз при максимуме или до тех пор, пока не будет выполнено условие take .

import * as rxCore from 'https://dev.jspm.io/rxjs@6/_esm2015/index';
import * as rxOps from 'https://dev.jspm.io/rxjs@6/_esm2015/operators';

const numberStream = rxCore.range(0, 10);
numberStream.pipe(
    rxOps.map(number => number * number),     //creates Observable with [0,1,4,9,16,25,36,49,64,81]
    rxOps.filter(number => number % 2 === 0), //creates Observable with [0,4,16,36,64]
    rxOps.take(3)         //completes after [0,4,16]
).subscribe(console.log); //logs 0, 4, 16

Есть ли какие-либо промежуточные наблюдаемые, создаваемые внутри этого конвейера операторов? Или только полный конвейер создает одну новую наблюдаемую область, оставляя numberStream нетронутым? Или в чем именно дело?

Answer 1

Почти каждый оператор Rx JS создает промежуточную наблюдаемую. На самом деле, я не нашел пример оператора, который не делает этого, поэтому я могу предположить, что все они делают.

Пример: map() внутренний оператор вызывает lift метод. lift делает то, что он создает новую Наблюдаемую и возвращает ее, объявляя текущую Наблюдаемую как source Наблюдаемая.

Это означает, что в вашем примере range создает одну Наблюдаемые, в то время как map, filter и take создают три новых наблюдаемых, каждая из которых имеет source Наблюдаемая от той, которая была создана до текущей. Итак, source Наблюдаемое для map становится созданным range и т. Д.

Итак, когда вы подписываетесь на любое Наблюдаемое, оно пытается для вызова source наблюдаемого оператора. В случае map, он вызывает , это call метод, который подписывается на источник Observable. В вашем случае он подпишется на тот, который создан range. Таким образом, вызов subscribe к источнику проходит вплоть до первого Observable (у которого нет source), что означает, что оператор не применяется.

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

Не беспокойтесь об этом. Промежуточные наблюдаемые - это просто JavaScript объекты. Они легкие, и вы не можете легко создать большие стеки этих объектов, которые могут вызвать проблемы с производительностью. Кроме этого, наблюдаемые являются ленивыми, они ничего не делают, если не происходит событие.

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

Ну, на мой взгляд, это наполовину правда. Да, они применяют операторы к каждому элементу, но не с помощью одного Observable, а путем создания нескольких. Возможно, так было до Rx JS 5, но не с сегодняшними версиями.

Answer 2

Рассматривая способ создания собственного пользовательского оператора для передачи pipe()

source=>source.pipe(...) 

, она ожидает, что это функция, которая возвращает наблюдаемое путем необязательного изменения / добавления / применения дополнительных операций к источник наблюдаемый. Я полагаю, что это не сразу заметно, так как в конце канала вам будет доступен только один источник.