Учебник RxJS: наблюдаемые, операторы и не только

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

Сегодня мы рассмотрим обзор реактивного программирования и RxJS и проведем вас через краткое руководство о том, как реализовать все фундаментальные компоненты RxJS в ваших приложениях.

Что такое реактивное программирование?

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

Реактивные программы структурированы вокруг событий, а не последовательного выполнения итеративного кода сверху вниз. Это позволяет им реагировать на триггерное событие независимо от того, на каком этапе находится программа.

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

Асинхронный против синхронного

Одна из основных концепций реактивного программирования — синхронные и асинхронные данные. Короче говоря, синхронные данные доставляются по одному в кратчайшие сроки.

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

Преимущества реактивного программирования

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

К другим преимуществам можно отнести:

• Высокая масштабируемость
• Чистый и читаемый
• Легко добавить поддержку нового события или ответа
• Улучшенный пользовательский интерфейс благодаря небольшому времени простоя

Реактивная парадигма также может быть объединена с другими парадигмами для формирования смеси, такой как объектно-ориентированное реактивное программирование (OORP) или функциональное реактивное программирование (FRP). Это смешиваемое качество делает реактивное программирование универсальной парадигмой, которую можно изменять для различных целей.

Что такое RxJS?

Реактивная парадигма доступна для многих языков через реактивные расширения или Rx-библиотеки. Эти библиотеки представляют собой загружаемые API-интерфейсы, которые добавляют поддержку основных реактивных инструментов, таких как наблюдатели и реактивные операторы. С помощью реактивных расширений разработчики могут преобразовывать обычные итеративные языки, такие как JavaScript, Python, C ++ и т. Д., В реактивные языки.

RxJS — это, в частности, инструмент функционального реактивного программирования с шаблоном наблюдателя и шаблоном итератора. Он также включает адаптированную форму функций массива JavaScript (сокращение, отображение и т. Д.) Для обработки асинхронных событий как коллекций.

Rx-библиотека JavaScript называется RxJS. RxJS стал очень популярным, потому что он упрощает реализацию асинхронных сценариев JavaScripts. Без расширений асинхронный JavaScript сложен в использовании и недостаточно развит. RxJS делает асинхронность более доступной с помощью инструментов, созданных специально для реактивного и асинхронного программирования.

Многие фреймворки веб-приложений, такие как Angular, основаны на структурах RxJS. В результате вы, возможно, уже косвенно использовали RxJS!

Далее мы разберем основные компоненты RxJS и покажем вам, как реализовать их в вашем собственном коде.

Наблюдаемые RxJS

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

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

Давайте посмотрим, как создать наблюдаемое!

const {Observable} = require(‘rxjs’)

const wrapArrayIntoObservable = arr => {

    return new Observable(subscriber => {

        for(let item of arr) {

            subscriber.next(item);

        }

    });

}

const data = [1, 2, 3, 4, 5];

const observable = wrapArrayIntoObservable(data);

observable.subscribe(val => console.log(‘Subscriber 1: ‘ + val));

observable.subscribe(val => console.log(‘Subscriber 2: ‘ + val));

// Output:
Subscriber1:1
Subscriber1:2
Subscriber1:3
Subscriber1:4
Subscriber1:5
Subscriber2:1
Subscriber2:2
Subscriber2:3
Subscriber2:4
Subscriber2:5

В строке 3 мы создаем wrapArrayIntoObservable()функцию, которая принимает массив в качестве параметра и превращает этот массив в файл observable. Затем эта функция передается Observableконструктору в строке 12 и запускается для каждого подписчика. Наконец, в строках 14 и 15 каждый подписчик распечатывает полученный поток данных.

Конвейер данных RxJS

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

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

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

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

const { from } = require(‘rxjs’);

const { tap, filter, map } = require(‘rxjs/operators’);

const arrayDataObservable$ = from([1, 2, 3, 4, 5]);

const dataPipeline = arrayDataObservable$.pipe(

    tap(val => console.log(‘Value passing through the stream: ‘ + val)),

    filter(val => val > 2),

    map(val => val * 2)

)

const subscribeToBaseObservable = subscriberName => {

    return arrayDataObservable$.subscribe(val => {

        console.log(subscriberName + ‘ received: ‘ + val);

    })

}

const subscribeToDataPipeline = subscriberName => {

    return dataPipeline.subscribe(val => {

        console.log(subscriberName + ‘ received: ‘ + val);

    })

}

const handleSubscriptionToBaseObservable = () => {

    const subscription1 = subscribeToBaseObservable(‘Subscriber1’);

    const subscription2 = subscribeToBaseObservable(‘Subscriber2’);

}

const handleSubscriptionToDataPipeline = () => {

    const subscription1 = subscribeToDataPipeline(‘Subscriber1’);

    const subscription2 = subscribeToDataPipeline(‘Subscriber2’);

}

// 1. Execute this function first

handleSubscriptionToBaseObservable();

// 2. Execute this function next

//handleSubscriptionToDataPipeline();

//raw output
Subscriber1 received:1
Subscriber1 received:2
Subscriber1 received:3
Subscriber1 received:4
Subscriber1 received:5
Subscriber2 received:1
Subscriber2 received:2
Subscriber2 received:3
Subscriber2 received:4
Subscriber2 received:5


//filtered output
Value passing through the stream:1
Value passing through the stream:2
Value passing through the stream:3
Subscriber1 received:6
Value passing through the stream:4
Subscriber1 received:8
Value passing through the stream:5
Subscriber1 received:10
Value passing through the stream:1
Value passing through the stream:2
Value passing through the stream:3
Subscriber2 received:6
Value passing through the stream:4
Subscriber2 received:8
Value passing through the stream:5
Subscriber2 received:10

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

Операторы создания RxJS

Наиболее распространенными операторами, используемыми в конвейерах данных RxJS, являются операторы создания. Мы рассмотрим простой fromоператор создания, использованный в предыдущем разделе, и тесно связанный с ним ofоператор.

from

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

Вот пример:

const { from } = require('rxjs');
 
 
const DATA_SOURCE = [ 'String 1', 'String 2', 'Yet another string', 'I am the last string' ];
const observable$ = from(DATA_SOURCE)
 
observable$.subscribe(console.log)

// output
String 1
String 2
Yet another string
I am the last string

of

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

const { of } = require('rxjs');


const DATA_SOURCE = [ 'String 1', 'String 2', 'Yet another string', 'I am the last string' ];
const observableArray$ = of(DATA_SOURCE)

console.log("Array data source")
observableArray$.subscribe(console.log)

console.log("\n")
console.log("Sequence data source")
const observableSequence$ = of('String 1', 'String 2', 'Yet another string', 'I am the last string')

observableSequence$.subscribe(console.log)

//output
Array data source
[ 'String 1',
  'String 2',
  'Yet another string',
  'I am the last string' ]


Sequence data source
String 1
String 2
Yet another string
I am the last string

Функция конвейера RxJS и конвейерные операторы

pipe()Функция вызывает все, кроме порождающих операторов операторов. Эти не создающие операторы являются операторами второго типа, называемыми конвейерными операторами.

Операторы конвейера принимают один наблюдаемый в качестве входных данных и возвращают наблюдаемый в качестве выходных данных, чтобы продолжить конвейер. Их можно вызывать как обычные функции, op1()(obs)но чаще они вызываются последовательно для формирования конвейера данных. pipe()Функция является экологически чистым способом вызвать несколько операторов в последовательности и, следовательно, предпочтительный способ вызова операторов.

// standard
op4()(op3()(op2()(op1()(obs))))

// pipe function
obs.pipe(
  op1(),
  op2(),
  op3(),
  op3(),
)

Рекомендуется использовать эту pipe()функцию, даже если вы вызываете только одного оператора.

Операторы фильтрации RxJS

Наиболее распространенный тип конвейерного оператора — это оператор фильтрации. Эти операторы удаляют все значения, не соответствующие переданным критериям. Мы рассмотрим популярные операторы filterи firstоператоры фильтрации.

filter

filterОператор принимает предикат функцию, как val => val + 1 == 3, которая применяется ко всем переданным значениям. Для каждого значения программа сравнивает данное значение с функцией предиката и сохраняет все значения, которые составляют функцию true.

В приведенном ниже примере разрешены только четные числа:

const { from } = require('rxjs');
const { filter } = require('rxjs/operators');
 
const observable$ = from([1, 2, 3, 4, 5, 6])
 
observable$.pipe(
    filter(val => val % 2 == 0)
).subscribe(console.log)

//output
2
4
6

filterОператор является отличным инструментом для преобразования данных в соответствии с потребностями абонентов конкретными. Например, некоторые пользователи могут захотеть увидеть все списки продуктов, в то время как другие могут захотеть увидеть только продукты определенного ценового диапазона.

first

firstОператор может использоваться двумя способами. По умолчанию он возвращает первое значение, испускаемое наблюдаемым. Преимущество возврата первого значения состоит в том, что время обработки очень мало, что делает это использование идеальным для случаев, когда достаточно простого и быстрого ответа.

const { from } = require('rxjs');
const { first } = require('rxjs/operators');

const observable$ = from([1, 2, 3, 4, 5, 6])

// take first
observable$.pipe(
    first()
).subscribe(console.log)

// output
1

Другое использование firstоператора добавляет функцию предиката или значение по умолчанию для сравнения с переданными значениями. Аналогично filter, firstзатем возвращает первое значение, соответствующее предикату. Это использование помогает выполнять поиск в потоке данных, когда вам нужно только одно значение.

const { from } = require('rxjs');
const { first } = require('rxjs/operators');

const observable$ = from([1, 2, 3, 4, 5, 6])

// Example 1 - take first that passes the predicate, or default otherwise
observable$.pipe(
    first(val => val > 6, -1)
).subscribe(console.log)

//output
-1

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

Что учить дальше

Поздравляем с завершением нашего краткого руководства по основам RxJS. Теперь у вас есть инструменты для создания наблюдаемых объектов, использования общих создающих и конвейерных операторов, объединения их всех вместе с pipe()функцией.

Но это только моментальный снимок того, что RxJS может вам предложить. Вот некоторые промежуточные темы, которыми вы можете заняться:

• Предметы
• Трансформационные и комбинационные операторы
• Пользовательские операторы
• Интеграция событий DOM
• Реактивная обработка ошибок