Асинхронное программирование в javascript (callback, promise, rxjs )

На практике: функциональное реактивное программирование на JavaScript

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

Для согласования всего, что вы, возможно, читали о FRP в JS, важно понять, что Rx, Bacon.js, Angular и другие не согласуются с двумя основными принципами определения FRP Конала Эллиота

Функциональное реактивное программирование в своей реализации в JavaScript относится к программированию в функциональном стиле при создании и реагировании на потоки. Это довольно далеко от оригинальной формулировки Эллиота (которая специально исключает потоки как компонент), но тем не менее вдохновляется традиционными FRP.

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

Без императивного или нечистого кода веб-приложение JS с пользовательским интерфейсом не было бы очень полезным, поскольку оно не могло бы взаимодействовать со своей средой.

Давайте взглянем на пример, чтобы продемонстрировать основные принципы FRP-вдохновленного JavaScript. Этот пример использует RxJS и печатает движения мыши в течение десяти секунд:

Вы можете проверить этот код в действии в JSFiddle: FRP-вдохновленном JavaScript. Запустите скрипт и, пока идет подсчет до 10, наведите указатель мыши в экран с результатом. Вы должны увидеть координаты мыши вместе со счетчиком. Тогда на экран выведется, где была ваша мышь во время каждого 1-секундного интервала времени.

Давайте кратко обсудим эту реализацию шаг за шагом.

Сначала мы создаем observable . Это таймер, добавляющий значение в коллекцию каждые (каждую секунду). Нам нужно обработать событие таймера, чтобы извлечь его значение и добавить к результирующему потоку.

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

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

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

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

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

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

Реализация: неизменяемость и изменяемость в JavaScript

В JavaScript строки и числовые литералы реализованы неизменяемыми. Это легко понять, если рассмотреть, как мы работаем с ними:

var str = 'Hello!';var anotherStr = str.substring(2);// результат: str = 'Hello!' (не изменена)// результат: anotherStr = 'llo!' (новая строка)

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

Числовые литералы также неизменяемы. Следующий пример всегда будет иметь одинаковый результат:

var three = 1 + 2;// результат: three = 3

Ни при каких обстоятельствах не может стать чем-либо, кроме .

Это демонстрирует, что в JavaScript присутствует реализация неизменяемости. Однако разработчики JS знают, что язык позволяет изменить многое. Например, объекты и массивы изменяемы. Рассмотрим следующий пример:

var arr = ;arr.push(4);// результат: arr = var obj = { greeting: 'Hello' };obj.name = 'Jon';// результат: obj = { greeting: 'Hello', name: 'Jon' }

В этих примерах исходные объекты изменены. Новые объекты не возвращаются.

Чтобы узнать больше об изменяемости в других языках, ознакомьтесь с Изменяемые и неизменяемые объекты.

Идем на сервер

Уже сейчас мы можем описать логику запроса и обработки ответа. Пока мы умеем только работать с promise. Поэтому опишем еще один оператор map, который будет вызывать метод getUsersRepsFromAPI. В наблюдателе опишем логику обработки нашего promise:

/* Код учебный! В рабочих проектах с RxJS лучше избегать использования promise, вместо него стоит использовать потоки */fromEvent(input, 'keyup').pipe(  debounceTime(700),  map(event => event.target.value),  filter(val => val.length > 2),  distinctUntilChanged(),  map(value => getUsersRepsFromAPI(value))).subscribe({  next: promise => promise.then(reps => recordRepsToList(reps))});

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

Конечно, мы можем навесить catch на promise в методе next, но из-за этого наш код начнет все больше напоминать “callback hell”. Если вдруг нам понадобится выполнить еще один запрос, то сложность кода возрастет.

Примечание: использование promise в коде с RxJS считается антипаттерном. Promise имеет множество недостатков по сравнению с observable. Его нельзя отменить, и нельзя использовать повторно. Если перед вами стоит выбор, то выбирайте observable. То же самое касается метода toPromise класса Observable. Данный метод был реализован в целях совместимости с библиотеками, которые не могут работать с потоками.

Мы можем использовать метод from, чтобы спроецировать promise на поток, но этот способ чреват дополнительными вызовами метода subscribe, и также приведет к разрастанию и усложнению кода.

Решить эту проблему можно с помощью оператора mergeMap:

fromEvent(input, 'keyup').pipe(  debounceTime(700),  map(event => event.target.value),  filter(val => val.length > 2),  distinctUntilChanged(),  mergeMap(value => from(getUsersRepsFromAPI(value)))).subscribe({  next: reps => recordRepsToList(reps),  error: console.log})

Теперь нам не нужно писать логику обработки promise. Метод from сделал из promise поток, а оператор mergeMap обработал его. Если promise выполнится успешно, то вызовется метод next, и наш наблюдатель получит готовый объект. Если же произойдет ошибка, то будет вызван метод error, и наш наблюдатель выведет ошибку в console.

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

Future и Promise

Future – ссылка на значение или код ошибки, которая может стать доступной (только для чтения) в какой-то момент времени.

Promise – соответствующий дескриптор с возможностью одиночной записи, который обеспечивает доступ к значению. Аналоги объектов Future и Promise реализованы в большинстве популярных языков: Java, C#, Ruby, Python и т.д.

Функция, которая возвращает результат асинхронно, создаёт объект Promise, инициализирует асинхронную обработку, устанавливает итоговую функцию обратного вызова, которая в какой-то момент наполнит Promise данными, и возвращает вызывающему компоненту объект Future, связанный с Promise. Затем вызывающий компонент может прикрепить к Future некий код, который будет выполнен при появлении значений в этом объекте.

Во всех реализациях Future предусмотрен механизм для превращения блока кода в новый объект Future так, чтобы он выполнил переданный ему код в другом потоке и поместил итоговый результат внутрь связанного с ним объекта Promise. Таким образом, Future предоставляет простой способ делать код асинхронным. Объекты Future возвращают либо результат успешных вычислений, либо ошибку.

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

Concepts

Degrees of explicitness

Reactive programming languages can range from very explicit ones where data flows are set up by using arrows, to implicit where the data flows are derived from language constructs that look similar to those of imperative or functional programming. For example, in implicitly lifted functional reactive programming (FRP) a function call might implicitly cause a node in a data flow graph to be constructed. Reactive programming libraries for dynamic languages (such as the Lisp «Cells» and Python «Trellis» libraries) can construct a dependency graph from runtime analysis of the values read during a function’s execution, allowing data flow specifications to be both implicit and dynamic.

Sometimes the term reactive programming refers to the architectural level of software engineering, where individual nodes in the data flow graph are ordinary programs that communicate with each other.

Static or dynamic

Reactive programming can be purely static where the data flows are set up statically, or be dynamic where the data flows can change during the execution of a program.

The use of data switches in the data flow graph could to some extent make a static data flow graph appear as dynamic, and blur the distinction slightly. True dynamic reactive programming however could use imperative programming to reconstruct the data flow graph.

Higher-order reactive programming

Reactive programming could be said to be of higher order if it supports the idea that data flows could be used to construct other data flows. That is, the resulting value out of a data flow is another data flow graph that is executed using the same evaluation model as the first.

Data flow differentiation

Ideally all data changes are propagated instantly, but this cannot be assured in practice. Instead it might be necessary to give different parts of the data flow graph different evaluation priorities. This can be called differentiated reactive programming.

For example, in a word processor the marking of spelling errors need not be totally in sync with the inserting of characters. Here differentiated reactive programming could potentially be used to give the spell checker lower priority, allowing it to be delayed while keeping other data-flows instantaneous.

However, such differentiation introduces additional design complexity. For example, deciding how to define the different data flow areas, and how to handle event passing between different data flow areas.

Evaluation models of reactive programming

Evaluation of reactive programs is not necessarily based on how stack based programming languages are evaluated. Instead, when some data is changed, the change is propagated to all data that is derived partially or completely from the data that was changed. This change propagation could be achieved in a number of ways, where perhaps the most natural way is an invalidate/lazy-revalidate scheme.

It could be problematic simply to naively propagate a change using a stack, because of potential exponential update complexity if the data structure has a certain shape. One such shape can be described as «repeated diamonds shape», and has the following structure:
A_n→B_n→A_n+1, A_n→C_n→A_n+1, where n=1,2… This problem could be overcome by propagating invalidation only when some data is not already invalidated, and later re-validate the data when needed using lazy evaluation.

One inherent problem for reactive programming is that most computations that would be evaluated and forgotten in a normal programming language, needs to be represented in the memory as data-structures.[citation needed] This could potentially make reactive programming highly memory consuming. However, research on what is called lowering could potentially overcome this problem.

On the other side, reactive programming is a form of what could be described as «explicit parallelism»[citation needed], and could therefore be beneficial for utilizing the power of parallel hardware.

Similarities with observer pattern

Reactive programming has principal similarities with the observer pattern commonly used in object-oriented programming. However, integrating the data flow concepts into the programming language would make it easier to express them and could therefore increase the granularity of the data flow graph. For example, the observer pattern commonly describes data-flows between whole objects/classes, whereas object-oriented reactive programming could target the members of objects/classes.

Пишем приложение

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

Требований будет немного:

• не выполнять запрос к API, если введенная в input строка содержит менее 3-х символов;
• Чтобы не выполнять запрос на каждый введенный пользователем символ, следует установить задержку(debounce) в 700 миллисекунд перед обращением к API;

Для поиска репозиториев мы воспользуемся github API. Сами же примеры я рекомендую выполнять на stackblitz. Там же я выложил готовую реализацию. Ссылки представлены в конце статьи.

Начнем с html разметки. Опишем input и ul элементы:

<input type="text"><ul></ul>

Затем в js или ts файле получим ссылки на текущие элементы используя browser API:

const input = document.querySelector('input');const ul = document.querySelector('ul');

Ещё нам понадобится метод, который будет выполнять запрос к github API. Ниже приведен код функции getUsersRepsFromAPI, которая принимает на вход никнейм пользователя и выполняет ajax запрос, используя fetch. Затем возвращает promise, попутно преобразуя успешный ответ в json:

const getUsersRepsFromAPI = (username) => {  const url = `https://api.github.com/users/${ username }/repos`;  return fetch(url)    .then(response => {      if(response.ok) {        return response.json();      }      throw new Error('Ошибка');    });}

Следом напишем метод, который будет выводить список названий репозиториев:

const recordRepsToList = (reps) => {  for (let i = 0; i < reps.length; i++) {    // если элемент не существует, то создаем его    if (!ul.children) {      const newEl = document.createElement('li');      ul.appendChild(newEl);    }    // записываем название репозитория в элемент    const li = ul.children;    li.innerHTML = reps.name;  }  // удаляем оставшиеся элементы  while (ul.children.length > reps.length) {    ul.removeChild(ul.lastChild);  }}

Приготовления завершены. Настало время посмотреть на RxJS в действии. Нам необходимо слушать событие keyup нашего input’а. Первым делом мы должны понять, что в реактивном подходе мы работаем с потоками. К счастью, в RxJS уже предусмотрен подобный вариант. Вспомните метод fromEvent, который я упоминал выше. Используем его:

const keyUp = fromEvent(input, 'keyup');keyUp.subscribe({  next: console.log});

Теперь наше событие представлено как поток. Если мы посмотрим, что выводится в консоль, то увидим объект типа KeyboardEvent. Но нам нужно введенное пользователем значение. Вот тут-то нам и пригодится метод pipe и оператор map:

fromEvent(input, 'keyup').pipe(  map(event => event.target.value)).subscribe({  next: console.log});

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

fromEvent(input, 'keyup').pipe(  map(event => event.target.value),  filter(value => value.length > 2))

С первым требованием разобрались. Приступим ко второму. Нам необходимо реализовать debounce. В RxJS есть оператор debounceTime. Данный оператор в качестве первого аргумента принимает число миллисекунд, в течение которых значение будет удерживаться, прежде чем пройдет дальше. При этом каждое новое значение будет сбрасывать таймер. Таким образом, на выходе мы получим последнее значение, после ввода которого прошло 700 миллисекунд.

fromEvent(input, 'keyup').pipe(  debounceTime(700),  map(event => event.target.value),  filter(value => value.length > 2))

Вот как может выглядеть наш поток без debounceTime:

А вот так будет выглядеть тот же поток, пропущенный через этот оператор:

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

В целях дополнительной оптимизации предлагаю использовать еще один оператор — distinctUntilChanged. Данный метод избавит нас от дубликатов. Лучше всего показать его работу на примере:

from('aaabccc').pipe(  distinctUntilChanged())

Без distinctUntilChanged:

С distinctUntilChanged:

На практике: изменяемость в JavaScript

Функциональное программирование в JavaScript хорошо развивается. Но по своей сущности JS — очень изменчивый язык, состоящий из множества парадигм. Ключевая особенность функционального программирования — неизменяемость. Другие функциональные языки выбросят ошибку, когда разработчик попытается изменить неизменяемый объект. Тогда как мы можем примирить врожденную изменяемость JS при написании функционального или функционального реактивного JS?

Когда мы говорим о функциональном программировании в JS, слово «неизменяемое» используется много, но разработчик обязан всегда держать ее в голове. Например, Redux полагается на одно неизменяемое дерево состояний. Однако сам JavaScript способен изменять объект состояния. Чтобы реализовать неизменяемое дерево состояний, нам нужно каждый раз при изменении состояния возвращать новый объект состояния.

Для неизменяемости объекты JavaScript также могут быть заморожены с помощью

Обратите внимание, что это «неглубокая» заморозка — значения объектов внутри замороженного объекта все еще могут быть изменены. Для гарантированной неизменяемости такие функции «глубокой» заморозки, как Mozilla deepFreeze() и npm deep-freeze могут рекурсивно замораживать объекты

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

Существуют также библиотеки, поддерживающие неизменяемость в JS. Mori предоставляет постоянные структуры данных на основе Clojure. Immutable.js от Facebook также предоставляет неизменяемые коллекции для JS. Библиотеки утилит, такие как Underscore.js и lodash, предоставляют методы и модули для более функционального стиля программирования (а стало быть направленного на неизменяемость).

RxAndroid

Порт библиотеки ReactiveX для мира Android называется rxAndroid и подключается, как всегда, через Gradle.

Издатель, генерирующий уведомления, здесь задается с помощью класса Observable. У издателя может быть несколько подписчиков, для их реализации воспользуемся классом Subscriber. Стандартное поведение для Observable — выпустить одно или несколько сообщений для подписчиков, а затем завершить свою работу или выдать сообщение об ошибке. В качестве сообщений могут быть как переменные, так и целые объекты.

В данном случае издатель myObserv сначала отправит строки hello и message, а затем сообщение об успешном завершении работы. Издатель может вызвать методы , и , поэтому у подписчиков они должны быть определены.

Все готово для работы. Осталось связать объекты между собой — и «Hello, world!» в реактивном программировании готов!

Надо сказать, что это был очень простой пример. В ReactiveX есть множество вариантов поведения всех участников паттерна: фильтрация, группирование, обработка ошибок. Пользу от реактивного программирования можно ощутить, только попробовав его в деле. Приступим к задаче посерьезнее.

Вариант 1. Присоединись к сообществу «Xakep.ru», чтобы читать все материалы на сайте

Членство в сообществе в течение указанного срока откроет тебе доступ ко ВСЕМ материалам «Хакера», увеличит личную накопительную скидку и позволит накапливать профессиональный рейтинг Xakep Score!
Подробнее

Вариант 2. Открой один материал

Заинтересовала статья, но нет возможности стать членом клуба «Xakep.ru»? Тогда этот вариант для тебя!
Обрати внимание: этот способ подходит только для статей, опубликованных более двух месяцев назад.

Я уже участник «Xakep.ru»

Слияние

merge()

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

Flux<User> mergeFluxWithInterleave(Flux<User> flux1, Flux<User> flux2) {
    return Flux.merge(flux1, flux2);
}

Вот два репозитория (весь код по выше уже приведенной ссылке):

// здесь два пользователя, отдаются сразу
ReactiveRepository<User> repository          = new ReactiveUserRepository(MARIE, MIKE); 
// а зедсь четыре, отдаются с задержкой 500 мс
ReactiveRepository<User> repositoryWithDelay = new ReactiveUserRepository(500);

merge() сольет потоки в итоговый поток, в котором пользователи идут в порядке следования во времени:

@Test
public void mergeWithInterleave() {
	Flux<User> flux = workshop.mergeFluxWithInterleave(repositoryWithDelay.findAll(), repository.findAll());
	StepVerifier.create(flux)
		    .expectNext(MARIE, MIKE, User.SKYLER, User.JESSE, User.WALTER, User.SAUL)
		    .verifyComplete();
}

concat()

Если нужно слить потоки так, чтобы сначала шли пользователи из первого потока, а потом из второго, независимо от того, как они приходят в реальном времени, то используйте concat():

Flux<User> mergeFluxWithNoInterleave(Flux<User> flux1, Flux<User> flux2) {
    return Flux.concat(flux1, flux2);
}

Проверим:

@Test
public void mergeWithNoInterleave() {
	Flux<User> flux = workshop.mergeFluxWithNoInterleave(repositoryWithDelay.findAll(), repository.findAll());
	StepVerifier.create(flux)
		.expectNext(User.SKYLER, User.JESSE, User.WALTER, User.SAUL, MARIE, MIKE)
		.verifyComplete();
}

С помощью concat() можно слить и потоки Mono:

Flux<User> createFluxFromMultipleMono(Mono<User> mono1, Mono<User> mono2) {
    return Flux.concat(mono1, mono2);
}

Заключение

Мы наконец-то перешли к практике и увидели, для чего нужны пайпы и операторы. Рассмотрели, как можно сократить код, пользуясь богатым API, которое предоставляет нам RxJS. Конечно, наше приложение не закончено, в следующей части мы разберем, как можно в одном потоке обрабатывать другой и как отменять наш http запрос, чтобы еще больше экономить трафик и ресурсы нашего приложения. А чтобы вы могли увидеть разницу, я выложил пример без использования RxJS, посмотреть его можно здесь. По этой ссылке вы найдете полный код текущего приложения. Для генерации схем я пользовался RxJS визуализатором.

Подводя итог: observables

Observables являются потоками. Мы можем наблюдать любой поток: от событий изменения размеров в существующих массивах до ответов API. Мы можем создать observables практически из всего, что угодно. Promise — тот же observable, но отдающий только одно значение, а observables могут возвращать много значений с течением времени.

Мы можем работать с observables по-разному. RxJS использует множество операторов. Observables часто визуализируется с помощью точек на линии, как показано на сайте RxMarbles. Поскольку поток состоит из асинхронных событий с течением времени, легко осмыслять это линейным способом и использовать именно такие зрительные образы, чтобы понять реактивные операторы. Например, следующие изображение от RxMarbles иллюстрирует :

Дополнительную информацию об observables можно найти на следующих ресурсах:

• Реактивные расширения: observables
• Создание и подписка на простые последовательности observable
• Представляем observable
• RxMarbles
• Rx Book — observable
• Представление observables

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

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

Реактивное программирование часто связано с Reactive Extensions, API для асинхронного программирования с observable потоками. Реактивные расширения (с префиксом Rx) предоставляют библиотеки для различных языков, включая JavaScript (RxJS).