Асинхронно програмиране в JavaScript (Callback, Promise, RxJs)

Здравейте всички. В контакт Омелницки Сергей. Не толкова отдавна бях домакин на поток за реактивно програмиране, където говорих за асинхронност в JavaScript. Днес бих искал да обобщя този материал.

Но преди да започнем основния материал, трябва да направим въведение. И така, нека започнем с определенията: какво са стек и опашка?

Пържола е колекция, чиито елементи се извличат на базата на LIFO „последен влязъл, първи излязъл“.

завъртете е колекция, чиито елементи се получават на принципа („първи влязъл, пръв излязъл“ FIFO

Добре, да продължим.

JavaScript е език за програмиране с една нишка. Това означава, че има само една нишка за изпълнение и един стек, където функциите са на опашка за изпълнение. Следователно JavaScript може да изпълнява само една операция в даден момент, докато други операции ще чакат своя ред в стека, докато бъдат извикани.

Стек от обаждания е структура от данни, която с прости думи записва информация за мястото в програмата, където се намираме. Ако скочим във функция, избутваме нейния запис до върха на стека. Когато се върнем от функция, изваждаме най-горния елемент от стека и завършваме там, откъдето сме извикали тази функция. Това е всичко, което стекът може да направи. И сега един много интересен въпрос. Как тогава работи асинхронността в JavasScript?

Всъщност, в допълнение към стека, браузърите имат специална опашка за работа с така наречения WebAPI. Функциите от тази опашка ще се изпълняват по ред само след като стекът е напълно изчистен. Едва след това те се поставят от опашката върху стека за изпълнение. Ако има поне един елемент в стека в момента, тогава те не могат да попаднат в стека. Точно поради това извикването на функции чрез таймаут често е неточно във времето, тъй като функцията не може да стигне от опашката до стека, докато е пълна.

Нека да разгледаме следния пример и да го разгледаме стъпка по стъпка. Да видим и какво се случва в системата.

console.log('Hi');
setTimeout(function cb1() {
    console.log('cb1');
}, 5000);
console.log('Bye');

1) Засега нищо не се случва. Конзолата на браузъра е чиста, стекът за повиквания е празен.

2) След това командата console.log('Hi') се добавя към стека на повикванията.

3) И е изпълнено

4) Тогава console.log('Hi') се премахва от стека на повикванията.

5) Сега нека преминем към командата setTimeout(function cb1() {… }). Добавя се към стека на повикванията.

6) Командата setTimeout(function cb1() {… }) се изпълнява. Браузърът създава таймер, който е част от уеб API. Ще извърши обратно броене.

7) Командата setTimeout(function cb1() {… }) е приключила работата си и е премахната от стека за извикване.

8) Командата console.log('Bye') се добавя към стека за повиквания.

9) Командата console.log('Bye') се изпълнява.

10) Командата console.log('Bye') се премахва от стека на повикванията.

11) След като са изминали поне 5000 ms, таймерът приключва и поставя обратното извикване cb1 в опашката за обратно извикване.

12) Цикълът за събития взема функция cb1 от опашката за обратно извикване и я избутва в стека за повиквания.

13) Функцията cb1 се изпълнява и добавя console.log('cb1') към стека на повикванията.

14) Командата console.log('cb1') се изпълнява.

15) Командата console.log('cb1') се премахва от стека на повикванията.

16) Функция cb1 е премахната от стека за извикване.

Нека да разгледаме пример в динамика:

Е, ние разгледахме как асинхронността се прилага в JavaScript. Сега нека поговорим накратко за еволюцията на асинхронния код.

Еволюцията на асинхронния код.

a(function (resultsFromA) {
    b(resultsFromA, function (resultsFromB) {
        c(resultsFromB, function (resultsFromC) {
            d(resultsFromC, function (resultsFromD) {
                e(resultsFromD, function (resultsFromE) {
                    f(resultsFromE, function (resultsFromF) {
                        console.log(resultsFromF);
                    })
                })
            })
        })
    })
});

Асинхронното програмиране, както го познаваме в JavaScript, може да се извършва само с функции. Те могат да бъдат предавани като всяка друга променлива на други функции. Така се родиха обратните повиквания. И е готино, забавно и пламенно, докато не премине в тъга, меланхолия и тъга. Защо? Да, просто е:

• С нарастването на сложността на кода проектът бързо се превръща в неясни множество вложени блокове - „ад за обратно извикване“.
• Обработката на грешки може лесно да бъде пренебрегната.
• Не можете да връщате изрази с return.

С появата на Promise ситуацията стана малко по-добра.

new Promise(function(resolve, reject) {
    setTimeout(() => resolve(1), 2000);

}).then((result) => {
    alert(result);
    return result + 2;

}).then((result) => {
    throw new Error('FAILED HERE');
    alert(result);
    return result + 2;

}).then((result) => {
    alert(result);
    return result + 2;

}).catch((e) => {
    console.log('error: ', e);
});

• Появиха се обещаващи вериги, които подобриха четимостта на кода
• Имаше отделен метод за прихващане на грешки
• Паралелно изпълнение с добавен Promise.all
• Можем да разрешим вложена асинхронност с async/await

Но обещанието има своите ограничения. Например, обещание, без танци с тамбура, не може да бъде отменено и най-важното е, че работи с една стойност.

Е, тук плавно наближаваме реактивното програмиране. Изморен? Е, хубавото е, че можете да отидете да сварите няколко чайки, да помислите и да се върнете, за да прочетете повече. И ще продължа.

Реактивно програмиране - програмна парадигма, фокусирана върху потоците от данни и разпространението на промените. Нека да разгледаме по-подробно какво е поток от данни.

// Получаем ссылку на элемент
const input = ducument.querySelector('input');

const eventsArray = [];

// Пушим каждое событие в массив eventsArray
input.addEventListener('keyup',
    event => eventsArray.push(event)
);

Нека си представим, че имаме поле за въвеждане. Създаваме масив и за всеки клавиш на входното събитие ще съхраняваме събитието в нашия масив. В същото време бих искал да отбележа, че нашият масив е сортиран по време, т.е. индексът на по-късните събития е по-голям от индекса на по-ранните събития. Такъв масив е опростен модел на поток от данни, но все още не е поток. За да може този масив безопасно да се нарече поток, той трябва да може по някакъв начин да информира абонатите, че в него са пристигнали нови данни. Така стигаме до определението за поток.

Поток от данни

const { interval } = Rx;
const { take } = RxOperators;

interval(1000).pipe(
    take(4)
)

Flow е масив от данни, сортирани по време, който може да показва, че данните са се променили. Сега си представете колко удобно става да пишете код, в който трябва да задействате няколко събития в различни части на кода за едно действие. Ние просто се абонираме за потока и той ще ни уведоми, когато настъпят промени. И библиотеката RxJs може да направи това.

RxJS е библиотека за работа с асинхронни и базирани на събития програми, използващи наблюдаеми последователности. Библиотеката предоставя основния тип забележим, няколко типа помощници (Наблюдатели, плановици, субекти) и оператори за работа със събития като с колекции (карта, филтриране, намаляване, всеки и подобни от JavaScript Array).

Нека разберем основните концепции на тази библиотека.

Наблюдаем, наблюдател, производител

Observable е първият базов тип, който ще разгледаме. Този клас съдържа основната част от изпълнението на RxJs. Той е свързан с наблюдаем поток, за който можете да се абонирате чрез метода за абониране.

Observable реализира спомагателен механизъм за създаване на актуализации, т.нар Наблюдател. Извиква се източник на стойности за наблюдател Производител. Може да бъде масив, итератор, уеб сокет, някакъв вид събитие и т.н. Така че можем да кажем, че observable е проводник между Producer и Observer.

Observable обработва три вида събития на Observer:

• следващ - нови данни
• грешка - грешка, ако последователността е прекъсната поради изключение. това събитие също предполага края на поредицата.
• завършен - сигнал за края на последователността. Това означава, че няма да има повече нови данни

Да видим демонстрация:

В началото ще обработим стойностите 1, 2, 3 и след 1 сек. получаваме 4 и приключваме нашата нишка.

Мисли на глас

И тогава разбрах, че е по-интересно да разказвам, отколкото да пиша за това. 😀

Абонамент

Когато се абонираме за поток, създаваме нов клас абонамент, което ни дава възможност да се отпишем с метода отпишете. Можем също да групираме абонаменти, използвайки метода добавете. Е, логично е да можем да разгрупираме нишки с помощта на премахване на. Методите за добавяне и премахване приемат различен абонамент като вход. Бих искал да отбележа, че когато се отписваме, ние се отписваме от всички дъщерни абонаменти, сякаш те също са извикали метода за отписване. Продължавай.

Видове потоци

HOT
СТУД

Производителят е създаден извън наблюдаемото
Производителят е създаден вътре в наблюдаемия

Данните се предават по време на създаването на наблюдаемото
Данните се предоставят в момента на абонамента.

Нуждаете се от повече логика, за да се отпишете
Нишката се прекратява сама

Използва връзка "един към много".
Използва връзка едно към едно

Всички абонаменти имат еднаква стойност
Абонаментите са независими

Данните могат да бъдат загубени, ако няма абонамент
Преиздава всички стойности на потока за нов абонамент

За да направя аналогия, бих си представил горещ поток като филм в кино. В какъв момент си дошъл, от този момент си започнал да гледаш. Бих сравнил студен поток с обаждане в тях. поддържа. Всеки обаждащ се слуша записа на телефонния секретар от началото до края, но можете да затворите с отписване.

Бих искал да отбележа, че има и така наречените топли потоци (срещам такова определение изключително рядко и само в чужди общности) - това е поток, който се трансформира от студен поток в горещ. Възниква въпросът - къде да се използва)) Ще дам пример от практиката.

Работя с Angular. Той активно използва rxjs. За да получа данни към сървъра, очаквам студен поток и използвам този поток в шаблона с помощта на asyncPipe. Ако използвам тази тръба няколко пъти, тогава, връщайки се към дефиницията на студен поток, всяка тръба ще изисква данни от сървъра, което е меко казано странно. И ако конвертирам студен поток в топъл, тогава заявката ще се случи веднъж.

Като цяло разбирането на типа потоци е доста трудно за начинаещи, но важно.

Операторите

return this.http.get(`${environment.apiUrl}/${this.apiUrl}/trade_companies`)
    .pipe(
        tap(({ data }: TradeCompanyList) => this.companies$$.next(cloneDeep(data))),
        map(({ data }: TradeCompanyList) => data)
    );

Операторите ни предоставят възможност за работа с потоци. Те помагат да се контролират събитията, протичащи в Наблюдаваното. Ще разгледаме няколко от най-популярните, а повече информация за операторите можете да намерите на връзките в полезна информация.

Оператори-на

Нека започнем с помощния оператор на. Той създава Observable въз основа на проста стойност.

Оператори-филтър

Операторът за филтриране, както подсказва името, филтрира сигнала на потока. Ако операторът върне истина, той прескача по-нататък.

Оператори - вземете

take - Взема стойността на броя излъчвания, след което потокът приключва.

Оператори-debounceTime

debounceTime - отхвърля излъчени стойности, които попадат в определения интервал от време между изходните данни - след изтичане на интервала от време, излъчва последната стойност.

const { Observable } = Rx;
const { debounceTime, take } = RxOperators;

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Испускаем значение раз в 1000мс
  setInterval(() => {
    observer.next(i++)
  }, 1000);

 // Испускаем значение раз в 1500мс
  setInterval(() => {
    observer.next(i++)
  }, 1500);
}).pipe(
  debounceTime(700),  // Ожидаем 700мс значения прежде чем обработать
  take(3)
);  

Оператори-takeWhile

Издава стойности, докато takeWhile върне false, след което се отписва от нишката.

const { Observable } = Rx;
const { debounceTime, takeWhile } = RxOperators;

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Испускаем значение раз в 1000мс
  setInterval(() => {
    observer.next(i++)
  }, 1000);
}).pipe(
  takeWhile( producer =>  producer < 5 )
);  

Оператори-combineLatest

Комбинираният оператор combineLatest е донякъде подобен на promise.all. Той комбинира множество потоци в един. След като всяка нишка направи поне едно излъчване, ние получаваме най-новите стойности от всяка като масив. Освен това, след всяко излъчване от комбинираните потоци, това ще даде нови стойности.

const { combineLatest, Observable } = Rx;
const { take } = RxOperators;

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Испускаем значение раз в 1000мс
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Испускаем значение раз в 750мс
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

combineLatest(observer_1, observer_2).pipe(take(5));

Оператори-zip

Zip - чака стойност от всеки поток и формира масив въз основа на тези стойности. Ако стойността не идва от никоя нишка, тогава групата няма да бъде формирана.

const { zip, Observable } = Rx;
const { take } = RxOperators;

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Испускаем значение раз в 1000мс
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Испускаем значение раз в 750
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

const observer_3 = Observable.create((observer) => {
  let i = 1;
  // Испускаем значение раз в 500
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
});

zip(observer_1, observer_2, observer_3).pipe(take(5));

Оператори - forkJoin

forkJoin също се присъединява към нишки, но излъчва стойност само когато всички нишки са завършени.

const { forkJoin, Observable } = Rx;
const { take } = RxOperators;

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Испускаем значение раз в 1000мс
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
}).pipe(take(3));

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Испускаем значение раз в 750
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
}).pipe(take(5));

const observer_3 = Observable.create((observer) => {
  let i = 1;
  // Испускаем значение раз в 500
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
}).pipe(take(4));

forkJoin(observer_1, observer_2, observer_3);

Оператори-карта

Операторът за преобразуване на карта преобразува излъчената стойност в нова.

const {  Observable } = Rx;
const { take, map } = RxOperators;

Observable.create((observer) => {
  let i = 1;
  // Испускаем значение раз в 1000мс
  setInterval(() => {
    observer.next(i++);
  }, 1000);
}).pipe(
  map(x => x * 10),
  take(3)
);

Оператори - споделяне, докосване

Операторът за докосване ви позволява да правите странични ефекти, тоест всякакви действия, които не влияят на последователността.

Операторът на споделената програма може да превърне студения поток в горещ.

Операторите са готови. Нека да преминем към темата.

Мисли на глас

И тогава отидох да пия чай. Писна ми от тези примери 😀

Предметно семейство

Темата на семейството е отличен пример за горещи теми. Тези класове са вид хибрид, който действа като наблюдаем и наблюдател едновременно. Тъй като темата е горещ поток, трябва да се отпишете от нея. Ако говорим за основните методи, това са:

• следващ - предаване на нови данни към потока
• грешка - грешка и прекратяване на нишка
• завършен - край на нишката
• subscribe - абонирайте се за поток
• unsubscribe - отписване от потока
• asObservable - трансформирайте се в наблюдател
• toPromise - трансформира се в обещание

Разпределете 4 5 вида предмети.

Мисли на глас

Казах 4 на потока, но се оказа, че са добавили още един. Както се казва, живей и се учи.

Проста тема new Subject()- най-простият вид теми. Създаден без параметри. Предава стойностите, които са дошли само след абонамента.

BehaviorSubject new BehaviorSubject( defaultData<T> ) - според мен най-често срещаният тип субекти. Входът приема стойността по подразбиране. Винаги запазва данните от последния брой, които се предават при абониране. Този клас също има полезен метод за стойност, който връща текущата стойност на потока.

ReplaySubject new ReplaySubject(bufferSize?: number, windowTime?: number) - По желание може да вземе като първи аргумент размера на буфера от стойности, които ще съхранява в себе си, и втория път, през който имаме нужда от промени.

асинхронен предмет new AsyncSubject() - нищо не се случва при абониране и стойността ще бъде върната само когато е завършена. Ще бъде върната само последната стойност на потока.

WebSocketSubject new WebSocketSubject(urlConfigOrSource: string | WebSocketSubjectConfig<T> | Observable<T>, destination?: Observer<T>) - Документацията мълчи за това и аз самият го виждам за първи път. Който знае какво прави, пишете, ние ще добавим.

уф. Е, обмислихме всичко, което исках да кажа днес. Надяваме се, че тази информация е била полезна. Можете да прочетете сами списъка с литература в раздела Полезна информация.

Полезна информация

• Връзка към потока
• Как работи JS: преглед на двигателя, механизми за изпълнение, стек за повиквания
• Как работи JS: Цикълът на събитията, Async и пет начина за подобряване на вашия код с async / await
• Как работи цикълът на събитията в JavaSript
• Еволюцията на асинхронния JavaScript
• Какво е RxJS и защо е полезно да знаете за него
  
• Практическо приложение на RxJS
• Урок за RxJS Observables — Създаване и абониране за Observables
• RXJS: Горещи и студени наблюдаеми
• Класове, функции за създаване на Observable. Оператори.
• RxJS оператори чрез пример
• API списък
• Вариации на предмет и график в RxJS

Източник: www.habr.com