🥇Asynchronní programování v JavaScriptu. ( Callback, Promise, RxJs )

Ahoj všem. Tady je Sergej Omelnytsky. Nedávno jsem vedl stream o reaktivním programování, kde jsem hovořil o asynchronnosti v JavaScriptu. Dnes bych chtěl shrnout tento materiál.

Ale předtím, než začneme s hlavním materiálem, je potřeba udělat úvod. Tak pojďme začít s definicemi: co je to zásobník a fronta?
Zásobník — to je kolekce, jejíž prvky jsou vybírány podle principu „poslední přišel, první odešel“ LIFO.
Fronta — to je kolekce, jejíž prvky jsou vybírány podle principu „první přišel, první odešel“ FIFO.

Dobře, pokračujme.

JavaScript je jednoproduktový programovací jazyk. To znamená, že má pouze jeden vykonávací vláknový tok a jeden zásobník, do kterého se funkce ukládají do fronty k vykonání. Proto může JavaScript v jeden okamžik vykonat pouze jednu operaci, jiné operace na to čekají ve frontě, dokud nejsou vyvolány.

Zásobník volání — to je datová struktura, která, zjednodušeně řečeno, zapisuje informace o místě v programu, kde se nacházíme. Když přejdeme do funkce, umístíme záznam o ní na vrchol zásobníku. Když se z funkce vracíme, vytáhneme záznam z vrchní části zásobníku a ocitneme se tam, odkud byla tato funkce vyvolána. To je vše, co zásobník dělá. A teď vzrušující otázka. Jak tedy funguje asynchronnost v JavaScriptu?

Ve skutečnosti, kromě zásobníku, mají prohlížeče speciální frontu pro práci s takzvaným WebAPI. Funkce z této fronty se vykonávají v pořadí, až když je zásobník úplně vyprázdněn. Až pak jsou umístěny do zásobníku k vykonání. Pokud je v zásobníku v současnosti alespoň jeden prvek, nemohou do zásobníku vstoupit. Právě z tohoto důvodu bývá volání funkcí pomocí timeru často časově nepřesné, protože funkce nemůže vstoupit do zásobníku, dokud je ten plný.

Podívejme se na následující příklad a zabývejme se jeho krokovým „vykonáním“. Také se podívejme, co se v systému současně děje.

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

1) Zatím se nic neděje. Konzole prohlížeče je prázdná, zásobník volání je prázdný.

2) Poté se příkaz console.log(‘Ahoj’) přidá do zásobníku volání.

3) A ona se provádí

4) Poté se console.log(‘Ahoj’) odstraní ze zásobníku volání.

5) Nyní přejdeme k příkazu setTimeout(function cb1() {… }). Ten je přidán do zásobníku volání.

6) Příkaz setTimeout(function cb1() {… }) se provádí. Prohlížeč vytvoří časovač, který je součástí Web API. Bude provádět odpočet.

7) Příkaz setTimeout(function cb1() {… }) dokončil svou činnost a je odstraněn ze zásobníku volání.

8) Příkaz console.log(‘Sbohem’) je přidán do zásobníku volání.

9) Příkaz console.log(‘Sbohem’) se provádí.

10) Příkaz console.log(‘Sbohem’) je odstraněn ze zásobníku volání.

11) Jakmile uplyne alespoň 5000 ms, časovač dokončí svou činnost a umístí callback cb1 do fronty callbacků.

12) Cyklus událostí vezme funkci cb1 z fronty callbacků a umístí ji do zásobníku volání.

13) Funkce cb1 se provádí a přidává console.log(‘cb1’) do zásobníku volání.

14) Příkaz console.log(‘cb1’) se provádí.

15) Příkaz console.log(‘cb1’) je odstraněn ze zásobníku volání.

16) Funkce cb1 je odstraněna ze zásobníku volání.

Podívejme se na příklad v dynamice:

Tak a máme za sebou, jak je v JavaScriptu realizována asynchronnost. Teď si stručně promluvme o evoluci asynchronního kódu.

Evoluce asynchronního kódu.

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);
                    })
                })
            })
        })
    })
});

Asynchronní programování, jak ho známe v JavaScriptu, může být realizováno pouze pomocí funkcí. Ty mohou být předávány jako jakákoliv jiná proměnná dalším funkcím. Tak se zrodily callbacky. A je to skvělé, zábavné a veselé, dokud se to nepromění v smutek, lítost a zármutek. Proč? Odpověď je jednoduchá:

S rostoucí složitostí kódu se projekt rychle mění v málo srozumitelné vícenásobně vnořené bloky — „callback hell“.
Zpracování chyb lze snadno opomenout.
Není možné vracet výrazy pomocí return.

Se vznikem Promise se situace o něco zlepšila.

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

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

}).then((result) => {
    throw new Error('TADY DOŠLO K CHYBĚ');
    alert(result);
    return result + 2;

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

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

Objevily se řetězce promissů, což zlepšilo čitelnost kódu.
Objevil se samostatný metod pro zachycení chyb.
Objevila se možnost paralelního zpracování pomocí Promise.all
Vnořenou asynchronnost můžeme vyřešit pomocí async/await

Ale má svá omezení. Například promisi nelze zrušit bez zbytečných komplikací, a co je nejdůležitější – pracuje s jednou hodnotou.

Tak jsme se pomalu dostali k reaktivnímu programování. Jste unavení? Naštěstí si můžete udělat čaj, promyslet to a vrátit se k dalšímu čtení. Já budu pokračovat.

Reaktivní programování — paradigma programování zaměřená na toky dat a šíření změn. Pojďme si podrobněji vysvětlit, co znamená tok dat.

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

const eventsArray = [];

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

Představme si, že máme pole pro vstup. Vytváříme pole a na každou událost keyup ze vstupu budeme ukládat událost do našeho pole. Rádi bychom poznamenali, že naše pole je seřazeno podle času, tedy index pozdějších událostí je větší než index dřívějších. Takové pole představuje zjednodušenou model toku dat, ale zatím to není tok. Aby toto pole mohlo být označeno jako tok, musí mít nějaký způsob, jak zasílat novinky odběratelům o nových datech. Tímto jsme se dostali k definici toku.

Tok dat

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

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

Tok — to je pole dat seřazených podle času, které může informovat o změnách dat. A teď si představte, jak snadné je psát kód, kde je potřeba vyvolat několik událostí na různých místech kódu při jedné akci. Jednoduše se přihlásíme k toku a ten nám sám sdělí, kdy dojde ke změnám. A to zvládá knihovna RxJs.

RxJS — je knihovna pro práci s asynchronními a událostmi řízenými programy pomocí pozorovacích sekvencí. Knihovna poskytuje základní typ Observable, několik pomocných typů (Observer, Schedulers, Subjects) a operátory práce s událostmi jako s kolekcemi (map, filter, reduce, every a podobné z JavaScript Array).

Pojďme se zaměřit na základní pojmy této knihovny.

Observable, Observer, Producer

Observable je první základní typ, který budeme zkoumat. Tato třída obsahuje hlavní část implementace RxJs. Je spojena s pozorovaným tokem, na který se lze přihlásit pomocí metody subscribe.

V Observable je implementován pomocný mechanismus pro vytváření aktualizací, takzvaný Observer. Zdrojem hodnot pro Observer je to, co se nazývá Producent. Může to být pole, iterátor, web socket, nějaká událost atd. Lze tedy říci, že observable je prostředníkem mezi Producer a Observer.

Observable zpracovává tři typy událostí Observer:

next – nové údaje
error – chybu, pokud sekvence skončila z důvodu výjimečné situace. Tato událost rovněž předpokládá ukončení sekvence.
complete — signál o dokončení sekvence. To znamená, že již nebudou žádná nová data.

Podívejme se na demo:

Na začátku zpracujeme hodnoty 1, 2, 3 a po 1 sekunde získáme 4 a ukončíme náš tok.

Myšlenky nahlas

A tehdy jsem si uvědomil, že vyprávění bylo zajímavější než o tom psát. 😀

Subscription

Když se přihlásíme na tok, vytváříme novou třídu subscription, která nám umožňuje zrušit odběr pomocí metody unsubscribe. Také můžeme seskupit odběry pomocí metody add. A logicky, můžeme oddělit toky pomocí remove. Metody add a remove přijímají jiný odběr. Rád bych zdůraznil, že když se odhlašujeme, odhlašujeme se od všech podřízených odběrů, jako bychom u nich vyvolávali metodu unsubscribe. Pokračujme dál.

Typy toků

HOT	COLD
Producer je vytvářen zvenčí observable	Producer je vytvářen uvnitř observable
Údaje jsou předávány v okamžiku vytvoření observable	Údaje jsou sdělovány v okamžiku přihlášení
Je potřeba dodatečná logika pro odhlášení	Tok se ukončuje sám
Využívá vazbu jedna k mnoha	Využívá vazbu jedna k jedné
Všechny odběry mají jednotnou hodnotu	Odběry jsou nezávislé
Údaje mohou být ztraceny, pokud není odběr	Znovu vydává všechny hodnoty toku pro nový odběr

Pokud bych měl učinit analogii, představil bych si horký tok jako film v kině. V jakém okamžiku jsi přišel, od toho okamžiku začínáš sledovat. Studený tok bych přirovnal k telefonátu na technickou podporu. Každý, kdo zavolá, poslouchá nahrávku automatického odpovídače od začátku do konce, ale můžeš hovor ukončit pomocí unsubscribe.

Rád bych poznamenal, že existují také takzvané warm toky (toto označení jsem slyšel velmi zřídka a pouze v zahraničních komunitách) — je to tok, který se transformuje z chladného toku na horký. Vzniká otázka — kde ho použít)) Uvedu příklad z praxe.

Pracuji s Angularem. Aktivně využívá rxjs. Pro získávání dat ze serveru očekávám chladný tok a tento tok používám v šabloně pomocí asyncPipe. Pokud tento pipe používám několikrát, vrátím se k definici chladného toku, každý pipe bude požadovat data ze serveru, což je, mírně řečeno, divné. A pokud chladný tok převedu na teplý, bude požadavek proveden pouze jednou.

Obecně je pochopení typů toků pro začátečníky poměrně složité, ale důležité.

Operátory

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

Možnost rozšíření práce s toky nám poskytují operátory. Pomáhají kontrolovat události, které probíhají v Observable. Podíváme se na několik nejpopulárnějších, podrobněji se s operátory můžete seznámit prostřednictvím odkazů v užitečných informacích.

Operátory — of

Začněme pomocným operátorem of. Vytváří Observable na základě jednoduché hodnoty.

Operátory — filter

Filtrační operátor filter, jak lze pochopit podle názvu, filtruje signál toku. Pokud operátor vrátí pravdu, tak ho propouští dále.

Operátory — take

take — Přijímá hodnotu počtu emitovaných, po kterém skončí tok.

Operátory — debounceTime

debounceTime — odstraňuje emitované hodnoty, které padnou do určeného časového intervalu mezi výstupy — po uplynutí časového intervalu emití poslední hodnotu.

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

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Emitujeme hodnotu jednou za 1000ms
  setInterval(() => {
    observer.next(i++)
  }, 1000);

 // Emitujeme hodnotu jednou za 1500ms
  setInterval(() => {
    observer.next(i++)
  }, 1500);
}).pipe(
  debounceTime(700),  // Čekáme 700ms na hodnoty před jejich zpracováním
  take(3)
);

Operátory — takeWhile

Emití hodnoty, dokud takeWhile nevrátí false, poté se odhlásí z toku.

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

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Vydává hodnotu každých 1000ms
  setInterval(() => {
    observer.next(i++)
  }, 1000);
}).pipe(
  takeWhile( producer =>  producer < 5 )
);

Operátory — combineLatest

Kombinovaný operátor combineLatest je podobný promise.all. Spojuje několik proudů do jednoho. Jakmile každý proud vydá alespoň jeden emit, získáme poslední hodnoty od každého ve formě pole. Následně po jakémkoli emitu z kombinovaných proudů bude vracet nové hodnoty.

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Vydává hodnotu každých 1000ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Vydává hodnotu každých 750ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

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

Operátory — zip

Zip čeká na hodnotu z každého proudu a vytvoří pole na základě těchto hodnot. Pokud hodnota nepřijde z nějakého proudu, skupina nebude vytvořena.

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Vydává hodnotu každých 1000ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Vydává hodnotu každých 750ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

const observer_3 = Observable.create((observer) => {
  let i = 1;
  // Vydává hodnotu každých 500ms
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
});

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

Operátory — forkJoin

forkJoin také spojuje proudy, ale vydává hodnotu pouze tehdy, když jsou všechny proudy dokončeny (complete).

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Vydává hodnotu každých 1000ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
}).pipe(take(3));

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Vydává hodnotu každých 750ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
}).pipe(take(5));

const observer_3 = Observable.create((observer) => {
  let i = 1;
  // Vydává hodnotu každých 500ms
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
}).pipe(take(4));

forkJoin(observer_1, observer_2, observer_3);

Operátory — map

Transformační operátor map převádí hodnotu emitu na novou.

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

Observable.create((observer) => {
  let i = 1;
  // Vydává hodnotu každých 1000ms
  setInterval(() => {
    observer.next(i++);
  }, 1000);
}).pipe(
  map(x => x * 10),
  take(3)
);

Operátory – share, tap

Operátor tap umožňuje provádět vedlejší efekty, tedy nějaké akce, které neovlivňují sekvenci.

Uživatelský operátor share dokáže z chladného toku udělat horký.

S operátory jsme skončili. Přejdeme k Subject.

Myšlenky nahlas

A tady jsem šel na čaj. Unavily mě tyto příklady 😀

Rodina subjectů

Rodina subjectů je jasným příkladem horkých toků. Tyto třídy jsou jakýmsi hybridem, který funguje zároveň jako observable a observer. Protože subject je horkým tokem, je nutné se od něj odhlásit. Hlavními metodami jsou:

next – předání nových dat do toku
error – chyba a dokončení toku
complete – dokončení toku
subscribe – přihlásit se k toku
unsubscribe – odhlásit se od toku
asObservable – transformujeme na pozorovatele
toPromise – transformuje na promise

Rozlišuje se 4–5 typů subjectů.

Myšlenky nahlas

Na streamu jsem mluvil o 4, ale zjistil jsem, že přidali ještě jeden. Jak se říká, ve věku žij a ve věku se uč.

Jednoduchý Subject new Subject()– nejjednodušší typ subjectů. Vytváří se bez parametrů. Předává hodnoty přicházející až po přihlášení.

BehaviorSubject new BehaviorSubject(defaultData) – podle mého názoru nejrozšířenější typ subjectů. Přijímá hodnotu jako výchozí. Vždy uchovává data posledního emit, které předává při přihlášení. Tato třída má také užitečnou metodu value, která vrací aktuální hodnotu toku.

ReplaySubject new ReplaySubject(bufferSize?: number, windowTime?: number) – může jako první argument volitelně přijmout velikost vyrovnávací paměti hodnot, které bude uchovávat, a jako druhý čas, po který potřebujeme změny.

AsyncSubject new AsyncSubject() – při přihlášení se nic neděje a hodnota bude vrácena pouze při complete. Bude vrácena pouze poslední hodnota toku.

WebSocketSubject new WebSocketSubject(urlConfigOrSource: string | WebSocketSubjectConfig | Observable, destination?: Observer) – Dokumentace o něm mlčí a já ho vidím poprvé. Kdo ví, co dělá, napište, doplníme.

Uf. Tak jsme projeli vše, co jsem chtěl dnes říct. Doufám, že tyto informace byly užitečné. Seznam literatury si můžete prohlédnout na záložce užitečné informace.