Здравейте на всички. Аз съм Омелницки Сергей. Неотдавна проведох стрийм по реактивно програмиране, където говорих за асинхронността в JavaScript. Днес искам да обобщя този материал.
Но преди да започнем с основния материал, трябва да направим въвеждане. И така, нека започнем с определения: какво е стек и опашка?
Стек — това е колекция, елементите на която се получават по принципа „последен постъпил, първи излязъл“ LIFO
Опашка — това е колекция, елементите на която се получават по принципа „първи постъпил, първи излязъл“ FIFO
Добре, продължаваме.
JavaScript е еднопоточен език за програмиране. Това означава, че има само един поток на изпълнение и един стек, в който се поставят функциите в опашка за изпълнение. Следователно, в един момент време JavaScript може да изпълни само една операция, а другите операции ще чакат своя ред в стека, докато не бъдат извикани.
Стек от извиквания — това е структура от данни, която, опростено казано, записва информация за мястото в програмата, на което се намираме. Когато влизаме в функция, поставяме записа за нея в горната част на стека. Когато се връщаме от функцията, изтегляме от стека най-горния елемент и се оказваме там, откъдето сме извикали тази функция. Това е всичко, което стекът умее. А сега натискаме много интересен въпрос. Как тогава работи асинхронността в JavaScript?
Всъщност, освен стека в браузърите има специална опашка за работа с така нареченото 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() {… }) се изпълнява. Браузърът създава таймер, който е част от Web API. Той ще извърши обратно броене на времето.
7) Командата setTimeout(function cb1() {… }) завършва работа и се изтрива от стека на повикванията.
8) Командата console.log(‘Bye’) се добавя в стека на повикванията.
9) Командата console.log(‘Bye’) се изпълнява.
10) Командата console.log(‘Bye’) се изтрива от стека на повикванията.
11) След като изминат поне 5000 мс., таймерът завършва работа и слага колбек 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, може да бъде реализирано само с функции. Те могат да бъдат предавани както всяка друга променлива на други функции. Така се родиха колбековете. И е забавно, весело и интересно, докато не се превърне в тъга, мъка и скръб. Защо? Всичко е просто:
- С покачването на сложността на кода, проектът бързо се превръща в трудно разбираеми многократно вложени блокове — „callback hell“.
- Обработката на грешки лесно може да се пропусне.
- Невъзможно е да се върнат изрази с 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)
);Нека си представим, че имаме поле за въвеждане. Създаваме масив и на всяко събитие keyup на input ще запазим събитието в нашия масив. Следва да отбележим, че нашият масив е сортиран по време, т.е. индексът на по-късните събития е по-голям от индекса на по-ранните. Такъв масив представлява опростена модел на поток от данни, но все още не е поток. За да може този масив уверено да бъде наречен поток, той трябва да бъде способен по някакъв начин да съобщава на абонатите, че са постъпили нови данни. Така стигнахме до определението на потока.
Поток от данни
const { interval1 } = Rx;
const { take } = RxOperators;
interval(1000).pipe(
take(4)
)
Поток — това е масив от данни, сортиран по време, който може да съобщава, когато данните са се променили. А сега си представете колко удобно става да пишете код, в който за едно действие е необходимо да се извикат няколко събития на различни места в кода. Просто се абонираме за потока и той сам ще ни съобщи, когато настъпят промени. И библиотеката RxJs може да прави това.
RxJS — това е библиотека за работа с асинхронни и основани на събития програми, използвайки наблюдаеми последователности. Библиотеката предоставя основен тип Observable, няколко помощни типа (Observer, Schedulers, Subjects) и оператори за работа с събития като колекции (map, filter, reduce, every и подобни от JavaScript Array).
Нека разгледаме основните понятия на тази библиотека.
Observable, Observer, Producer
Observable — първият базов тип, който ще разгледаме. Този клас съдържа основната част от реализацията на RxJs. Той е свързан с наблюдавания поток, на който може да се абонираме с метода subscribe.
В Observable се реализира помощен механизъм за създаване на обновления, така нареченото Observer. Източникът на стойности за Observer се нарича Producer. Това може да бъде масив, итератор, web socket, някакво събитие и т.н. Така че можем да кажем, че observable е проводник между Producer и Observer.
Observable обработва три вида събития за Observer:
- next – нови данни
- error – грешка, ако последователността приключи поради изключителна ситуация. Това събитие също предполага завършване на последователността.
- complete — сигнал за завършване на последователността. Това означава, че нови данни повече няма да има
Нека видим демо:
В началото ще обработим стойностите 1, 2, 3, а след 1 сек. ще получим 4 и ще приключим нашия поток.
Мисли на глас
И тук осъзнах, че разказването беше по-интересно, отколкото да пиша за това. 😀
Subscription
Когато се абонираме за поток, създаваме нов клас subscription, който ни дава възможност да отменим абонамента чрез метода unsubscribe. Също можем да групираме абонаменти чрез метода add. И логично е, че можем да разгрупираме потокове с помощта на премахни. Методите add и remove приемат друга абонамент на входа. Бих искал да отбележа, че когато се отписваме, отписваме се от всички дъщерни абонаменти, сякаш сме извикали метода unsubscribe и за тях. Да продължим напред.
Видове потоци
| HOT | COLD |
|---|---|
| Producer се създава извън observable | Producer се създава вътре в observable |
| Данните се предават в момента на създаване на observable | Данните се предават в момента на абонамент |
| Необходима е допълнителна логика за отписване | Потокът завършва самостоятелно |
| Използва връзка един-к-многи | Използва връзка един-к-един |
| Всички абонаменти имат едно и също значение | Абонаментите са независими |
| Данните могат да се загубят, ако няма абонамент | Повторно издава всички стойности на потока за нов абонамент |
Ако трябва да направя аналогия, бих представил горещ поток като филм в кино. В момента, в който си дошъл, от там започваш да гледаш. Хладният поток бих сравнил със званен в техническа поддръжка. Всеки, който се обади, слуша записа на автоотговора от начало до край, но можеш да затвориш телефона чрез unsubscribe.
Искам да отбележа, че съществуват и така наречените топли потоци (такова определение съм срещал изключително рядко и само в чуждестранни общности) — това е поток, който се трансформира от студен в горещ. Възниква въпросът — къде да се използва)) Ще дам пример от практиката.
Работя с 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. Ще разгледаме няколко от най-популярните, а по-подробно за операторите може да се информирате чрез линковете в полезната информация.
Оператори — of
Започваме с помощния оператор of. Той създава Observable на базата на просто значение.
Оператори — filter
Операторът за филтриране filter, както може да се разбере от името, филтрира сигнала на потока. Ако операторът върне истина, тогава пропуска напред.
Оператори — take
take — Приймa стойност на броя на емитите, след което завършва потока.
Оператори — 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);
Оператори — map
Операторът за трансформация map преобразува стойността на изпускането в нова.
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)
);
Оператори – share, tap
Оператор tap позволява да се извършват странични ефекти, т.е. действия, които не влияят на последователността.
Утилитният оператор share може да превърне студен поток в горещ.
С операторите приключихме. Преминаваме към Subject.
Мисли на глас
И тук отидох да пия чай. Тези примери ме измориха 😀
Семейство на subject-ите
Семейството на subject-ите е ярък пример за горещи потоци. Тези класове представляват нещо като хибрид, който изпълнява едновременно ролите на observable и observer. Тъй като subject е горещ поток, от него трябва да се отписваме. Основните методи са:
- next – предаване на нови данни в потока
- error – грешка и приключване на потока
- complete – приключване на потока
- subscribe – абониране за потока
- unsubscribe – отписване от потока
- asObservable – трансформира в наблюдател
- toPromise – трансформира в промис
Изразяват се 4-5 типа subject-и.
Мисли на глас
На стрима казах 4, а се оказа, че те добавили още един. Както се казва, век живи, век учи.
Прост Subject new Subject()– най-простият вид subject-и. Създава се без параметри. Предава стойности, постъпили само след абонамент.
BehaviorSubject new BehaviorSubject( defaultData<T> ) – на мое мнение, най-разпространеният вид subject-и. Приема значение по подразбиране. Винаги запазва данните от последния emit, които предава при абонамент. Този клас има и полезния метод value, който връща текущата стойност на потока.
ReplaySubject new ReplaySubject(bufferSize?: number, windowTime?: number) – на вход опционално може да приеме първия параметър размер на буфера на стойностите, които ще съхранява, и втория параметър времето, в което ни трябват промените.
AsyncSubject new AsyncSubject() – при абонамент нищо не се случва и стойността ще бъде върната само при complete. Ще се върне само последната стойност на потока.
WebSocketSubject new WebSocketSubject(urlConfigOrSource: string | WebSocketSubjectConfig<T> | Observable<T>, destination?: Observer<T>) – за него документацията мълчи и аз го виждам за първи път. Ако някой знае какво прави, пишете, ще допълним.
Уф. Ето че разгледахме всичко, което исках да разкажа днес. Надявам се информацията да беше полезна. Самостоятелно можете да се запознаете със списъка с литература в раздела полезна информация.
Полезна информация
- – линк към стрима
- – Как работи JS: преглед на двигателя, механизмите за изпълнение, стека на извикванията
- — Как работи JS: цикъл на събития, асинхронност и пет начина за подобряване на кода с async / await
- — Как работи Event Loop в JavaScript
- — Еволюция на асинхронния JavaScript
- — Какво е RxJS и защо е полезно да знаете за него
- — Практическо приложение на RxJS
- — RxJS Observables Tutorial — Създаване & Подписване на Observables
- — RXJS: Горещи и Студени Observables
- — Класове и функции за създаване на Observable. Оператори.
- — Примери за RxJS Оператори
- — API Списък
- — Видове Subject и Графици в RxJS
Източник: habr.com
