Здравейте. Свързан е Омелницкий Сергей. Преди не много време проведох стрийм за реактивното програмиране, където говорих за асинхронността в 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 на входа ще запазваме събитието в нашия масив. Същевременно бих искал да подчертая, че нашият масив е подреден по време т.е. индексът на по-късните събития е по-голям от индекса на по-ранните. Такъв масив представлява опростен модел на поток от данни, но това все още не е поток. За да може този масив с увереност да бъде наречен поток, той трябва по някакъв начин да информира абонатите, че нови данни са влезли в него. По този начин стигнахме до определението за поток.
Данни поток
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 се нарича Производител. Това може да бъде масив, итератор, уеб сокет, някакво събитие и т.н. Така че можем да кажем, че observable е проводник между Producer и Observer.
Observable обработва три вида събития Observer:
- next – нови данни
- error – грешка, ако последователността приключи поради извънредна ситуация. това събитие също предполага приключване на последователността.
- complete — сигнал за приключване на последователността. Това означава, че нови данни вече няма да има
Нека разгледаме демонстрация:
В началото ще обработим стойностите 1, 2, 3, а след 1 сек. ще получим 4 и ще завършим потока ни.
Мисли на глас
И тук осъзнах, че е по-интересно да разказвам, отколкото да пиша за това. 😀
Subscription
Когато се абонираме за поток, създаваме нов клас subscription, който ни дава възможност да отменим абонамента с метода unsubscribe. Също така можем да групираме абонаменти с метода add. И логично, можем да разгрупираме потоковете чрез remove. Методите add и remove приемат друг абонамент като вход. Бих искал да подчертая, че когато отменяме абонамента, се отписваме от всички дъщерни абонаменти, сякаш също сме извикали метода unsubscribe за тях. Продължаваме напред.
Видове потоци
| HOT | COLD |
|---|---|
| Producer се създава извън observable | Producer се създава вътре в observable |
| Данните се предават в момента на създаване на observable | Данните се съобщават в момента на абонамента |
| Необходима е допълнителна логика за отписване | Потокът приключва самостоятелно |
| Използва връзка един-к-множество | Използва връзка един-к-един |
| Всички абонаменти имат едно и също значение | Абонаментите са независими |
| Данните могат да се загубят, ако няма абонамент | Преиздава всички стойности на потока за нов абонамент |
Ако трябва да дам аналогия, бих си представил горещия поток като филм в кино. Какъвто момент да дойдеш, от този момент и започваш гледането. Студения поток бих сравнил с обаждане до техническата поддръжка. Всеки, който се обади, слуша запис на автоотговора от началото до края, но можеш да затвориш телефона с помощта на unsubscribe.
Бих искал да отбележа, че съществуват така наречените топли потоци (такава дефиниция срещнах изключително рядко и само в чуждестранни общности) — това е поток, който се трансформира от студен поток в горещ. Изниква въпросът — къде да се използва)) Ще дам пример от практиката.
Работя с Angular. Той активно използва RxJS. За получаване на данни от сървъра очаквам студен поток и използвам този поток в шаблона с помощта на asyncPipe. Ако използвам този пайп няколко пъти, връщайки се към дефиницията на студения поток, всеки pipe ще търси данни от сървъра, което меко казано е странно. А ако преобразувам студения поток в топъл, заявката ще се случи само веднъж.
Въобще, разбирането на вида потоци е доста сложно за начинаещите, но е важно.
Оператори
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 — приема броя на емитите, след което завършва потока.
Оператори — 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 също комбинира потоци, но той емитира стойност само когато всички потоци са завършени (complete).
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-ове. Приема стойност по подразбиране. Винаги запазва данни от последния емит, които предава при абонамент. Този клас има и полезен метод 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 — Creating & Subscribing to Observables
- — RXJS: Горещи и студени Observables
- — Класове, функции за създаване на Observable. Оператори.
- — RxJS Оператори по пример
- — Списък на API
- — Разновидности Subject и Планиране в RxJS
Източник: habr.com
