🥇Aszinkron programozás a JavaScript-ben. (Callback, Promise, RxJs)

Helló mindenkinek. Omelnicki Szergej vagyok. Nemrégiben egy közvetítést tartottam az aszinkron programozásról, ahol a JavaScript aszinkronitásáról beszéltem. Ma szeretném összefoglalni ezt az anyagot.

De mielőtt elkezdenénk a fő anyagot, szükségünk van egy bevezetésre. Tehát kezdjük a definíciókkal: mi is az a verem és a sor?
Verem — ez egy olyan gyűjtemény, amelyből az elemek a 'legutolsó, aki belépett, az első, aki távozik' elv szerint kerülnek ki (LIFO).
Sor — ez egy olyan gyűjtemény, amelyből az elemek a 'legelső, aki belépett, az első, aki távozik' elv szerint kerülnek ki (FIFO).

Rendben, folytassuk.

A JavaScript egy egyszálú programozási nyelv. Ez azt jelenti, hogy csak egy végrehajtási szála és egy verem van, amelybe a funkciók a végrehajtásra várnak. Tehát egy adott időben a JavaScript csak egy műveletet hajthat végre, a többi művelet a veremben várakozik, amíg nem hívják őket.

Hívásverem — ez egy adatstruktúra, amely leegyszerűsítve feljegyzi a programban a helyet, ahol éppen vagyunk. Amikor egy funkcióba lépünk, feljegyként kerül a verem tetejére. Amikor visszatérünk a funkcióból, eltávolítjuk a veremből a legfelső elemet, és ott találjuk magunkat, ahonnan a funkciót hívtuk. Ez a verem egyetlen képessége. És most egy rendkívül érdekes kérdés. Hogyan működik akkor az aszinkronitás a JavaScript-ben?

Valójában a böngészőkben a verem mellett létezik egy külön sor, amely a WebAPI működéséhez szükséges. A sorban lévő funkciók végrehajtása azután történik meg sorrendben, hogy a verem teljesen kiürült. Csak ezután helyezhetőek át a sorból a verembe a végrehajtásra. Ha a veremben jelenleg van legalább egy elem, akkor az új elemek nem kerülhetnek a verembe. Ez az oka annak, hogy a timeout hívások gyakran pontatlan időben történnek, mivel a funkció nem tud átlépni a sorból a verembe, amíg az tele van.

Nézzük meg a következő példát és foglalkozzunk annak lépésről lépésre történő 'végrehajtásával'. Emellett nézzük meg, mi történik a rendszerben.

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

1) Jelenleg semmi sem történik. A böngésző konzolja tiszta, a hívásverem üres.

2) Ezután a console.log('Hi') parancs hozzáadódik a hívásveremhez.

3) És végrehajtódik.

4) Ezután a console.log('Hi') eltávolításra kerül a hívásveremből.

5) Most folytatjuk a setTimeout(function cb1() {… }) parancsra. Ez hozzáadódik a hívási veremhez.

6) A setTimeout(function cb1() {… }) parancs végrehajtódik. A böngésző létrehoz egy időzítőt, amely a Web API része. Ez elindítja az idő visszaszámlálását.

7) A setTimeout(function cb1() {… }) parancs befejeződik és eltávolítódik a hívási veremből.

8) A console.log(‘Bye’) parancs hozzáadódik a hívási veremhez.

9) A console.log(‘Bye’) parancs végrehajtódik.

10) A console.log(‘Bye’) parancs eltávolítódik a hívási veremből.

11) Miután eltelt legalább 5000 ms, az időzítő befejeződik és a cb1 visszahívást a visszahívások sorába helyezi.

12) Az eseményhurok kiveszi a cb1 függvényt a visszahívások sorából és a hívási verembe helyezi.

13) A cb1 függvény végrehajtódik és hozzáadja a console.log(‘cb1’) parancsot a hívási veremhez.

14) A console.log(‘cb1’) parancs végrehajtódik.

15) A console.log(‘cb1’) parancs eltávolítódik a hívási veremből.

16) A cb1 függvény eltávolítódik a hívási veremből.

Nézzük meg a példát dinamikusan:

Most pedig áttekintettük, hogyan valósul meg az aszinkronitás a JavaScript-ben. Beszéljünk röviden az aszinkron kód fejlődéséről.

Az aszinkron kód fejlődése.

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

Az aszinkron programozás, amit a JavaScript-ben ismerünk, csak függvényekkel valósítható meg. Ezek más függvényeknek bármely más változóként átadhatóak. Itt születtek a visszahívások. És ez szórakoztató, vidám, amíg meg nem fordul a dolog, és szomorúságba nem torkollik. Miért? Mert:

A kód bonyolultságának növekedésével a projekt gyorsan alig érthető, többszörösen beágyazott blokkokká válik - „callback hell”.
A hibakezelést könnyen el lehet nézni.
Kifejezéseket return-nel nem lehet visszaadni.

A Promise bevezetésével a helyzet valamivel javult.

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

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

}).then((result) => {
    throw new Error('Itt nem sikerült');
    alert(result);
    return result + 2;

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

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

Megjelentek a promise láncok, ami javította a kód olvashatóságát.
Megjelent egy külön hibakezelési módszer.
Lehetőség nyílt párhuzamos végrehajtásra a Promise.all segítségével.
A beágyazott aszinkronitást az async/await segítésével oldhatjuk meg.

A promissal kapcsolatban azonban vannak korlátai. Például egy promist anélkül, hogy táncikálnánk egy doboz körül, nem lehet lemondani, és ami a legfontosabb - csak egy értékkel működik.

Nos, így fokozatosan eljutottunk a reaktív programozásig. Fáradt vagy? Szerencsére lehet menni egy teát főzni, átgondolni, és visszatérni az olvasáshoz. Én folytatom.

Reaktív programozás — a programozás egy paradigmája, amely az adatok áramlására és a változások terjesztésére összpontosít. Nézzük meg részletesebben, mi az az adatfolyam.

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

const eventsArray = [];

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

Képzeljük el, hogy van egy bemeneti mezőnk. Létrehozunk egy tömböt, és minden keyup esemény során elmentjük az eseményt a tömbünkbe. Fontos megjegyezni, hogy a tömbünk időben rendezett, azaz a későbbi események indexe nagyobb, mint a korábbi eseményeké. Ez a tömb egy egyszerűsített adatfolyam modellt képvisel, de ez még nem adatfolyam. Ahhoz, hogy ezt a tömböt biztonsággal adatfolyamnak nevezzük, valamilyen módon tájékoztatnia kell a feliratkozókat arról, hogy új adatok érkeztek. Így jutottunk el az adatfolyam definíciójához.

Adatfolyam

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

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

Adatfolyam — ez egy időben rendezett adatok tömbje, amely képes jelezni, hogy az adatok megváltoztak. Most képzelje el, milyen kényelmes is lesz kódot írni, ahol egyetlen művelethez több eseményt kell hívni a kód különböző részein. Csak feliratkozunk az adatfolyamra, és ő magától jelzi, hogy mikor történnek változások. Ezt a RxJs könyvtár képes megtenni.

RxJS — ez egy könyvtár, amely aszinkron és eseményalapú programokkal dolgozik megfigyelt sorozatok használatával. A könyvtár a fő típust biztosítja, Observable, néhány segéd típust (Observer, Schedulers, Subjects) és operátorokat az események gyűjteményekként való kezelésére (map, filter, reduce, every és hasonlók a JavaScript tömbből).

Nézzük meg a könyvtár fő fogalmait.

Observable, Observer, Producer

Az Observable — az első alapvető típus, amit megvizsgálunk. Ez az osztály tartalmazza az RxJs főbb megvalósítását. Ez egy megfigyelt adatfolyammal van összekapcsolva, amelyre fel lehet iratkozni a subscribe metódussal.

Az Observable megvalósít egy segédmechanizmust a frissítések létrehozására, amit úgy nevezünk, Observer. Az Observer számára értékforrásnak nevezzük Producer. Ez lehet egy tömb, iterátor, web socket, valamilyen esemény stb. Így elmondható, hogy az observable egy közvetítő a Producer és az Observer között.

Az Observable háromféle eseményt kezel az Observer számára:

next – új adatok
error – hiba, ha a sorozat kivételi helyzet miatt ér véget. Ez az esemény is a sorozat befejezésére utal.
complete — jelez a sorozat befejezéséről. Ez azt jelenti, hogy új adatok már nem lesznek

Nézzük meg a demót:

Kezdetben az 1, 2, 3 értékeket dolgozzuk fel, majd 1 másodperc elteltével megkapjuk a 4-et, és befejezzük a folyamatot.

Gondolatok hangosan

És itt rájöttem, hogy érdekesebb volt mesélni, mint erről írni. 😀

Feliratkozás

Amikor feliratkozunk a folyamatra, egy új osztályt hozunk létre subscription, ami lehetővé teszi a feliratkozás lemondását a unsubscribemódszer segítségével. Csoportosíthatjuk a feliratkozásokat a addmódszerrel. Logikus, hogy a folyamatokat különböző előfizetések szerint is csoportosíthatjuk. remove. Az add és remove metódusok egy másik előfizetést várnak bemenetként. Érdemes megjegyezni, hogy amikor lemondunk a feliratkozásról, akkor az összes gyermeki előfizetést is lemondjuk, mintha az unsubscribe metódust hívtuk volna meg náluk is. Tovább lépünk.

A folyamatok típusai

HOT	COLD
A Producer kívülről jön létre az observable-be	A Producer belül jön létre az observable-ben
Az adatok az observable létrehozásakor kerülnek átadásra	Az adatok a feliratkozás pillanatában kerülnek átadásra
További logika szükséges a leiratkozáshoz	A folyamat önállóan ér véget
Egy-k-many kapcsolatot használ	Egy-k-egy kapcsolatot használ
Minden előfizetés közös értékkel bír	Az előfizetések függetlenek
Az adatok elveszhetnek, ha nincs előfizetés	Újra kibocsát minden adatot az új előfizetés számára

Hasonlatként a forró folyamatot úgy képzelem el, mint egy filmet a moziban. Amikor megérkeztél, attól a pillanattól kezded el a megtekintést. A hideg folyamatot a technikai támogatás hívásával hasonlítanám össze. Bárki, aki telefonál, a hangpostát hallgatja a legelejétől a végéig, de te leteheted a telefont az unsubscribe segítségével.

Érdemes megjegyezni, hogy léteznek úgynevezett meleg áramlások (ezt a kifejezést rendkívül ritkán találkoztam és csak külföldi közösségekben) — ez az áramlás, amely a hideg áramlásból forróvá alakul. Felmerül a kérdés — hol használjuk?)) Példát hozok a gyakorlatból.

Angularral dolgozom. Aktívan használja az rxjs-t. Az adatok szerverre történő lekérésénél hideg áramlást várok, és ezt az áramlást az asyncPipe segítségével használom a sablonban. Ha ezt a pipet többször használom, akkor a hideg áramlás definíciója szerint minden egyes pipe újra le fogja kérni az adatokat a szerverről, ami finoman szólva is furcsa. De ha átalakítom a hideg áramlást meleg áramlássá, akkor a lekérés egyszer történik.

Az áramlástípusok megértése általában elég nehéz a kezdők számára, de fontos.

Operátorok

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

Az áramlásokkal való munkavégzés lehetőségét az operátorok biztosítják. Segítenek ellenőrizni az Observable-ben zajló eseményeket. Megvizsgálunk néhány a legnépszerűbb közül, a részletesebb információkat az operátorokról a hasznos információk linkjein találhatja.

Operátorok — of

Kezdjük a segéd operátorral, az of-val. Ez egy Observable-t hoz létre egy egyszerű érték alapján.

Operátorok — filter

A filter szűrő operátor, ahogy a neve is sugallja, szűri az áramlás jeleit. Ha az operátor igazat ad vissza, akkor azt tovább engedi.

Operátorok — take

take — Elfogadja az emitált értékek számát, amely után befejezi az áramlást.

Operátorok — debounceTime

debounceTime — elveti azokat az emitált értékeket, amelyek a megadott időintervallumon belül érkeznek — az időintervallum leteltével emitálja az utolsó értéket.

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

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Emitálunk egy értéket 1000 ms-onként
  setInterval(() => {
    observer.next(i++)
  }, 1000);

 // Emitálunk egy értéket 1500 ms-onként
  setInterval(() => {
    observer.next(i++)
  }, 1500);
}).pipe(
  debounceTime(700),  // Várunk 700 ms-t az értékek feldolgozása előtt
  take(3)
);

Operátorok — takeWhile

Emitálja az értékeket, amíg a takeWhile nem ad vissza false-t, ezután leiratkozik az áramlásról.

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

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Érték kibocsátása 1000ms-onként
  setInterval(() => {
    observer.next(i++)
  }, 1000);
}).pipe(
  takeWhile( producer =>  producer < 5 )
);

Operátorok — combineLatest

A combineLatest operátor hasonló a promise.all-hoz. Egyesíti több streamet egybe. Miután minden stream legalább egyszer kibocsátott egy értéket, megkapjuk a legutolsó értékeket mindenből tömb formájában. Ezt követően pedig bármely kibocsátott érték esetén új értékeket ad vissza a kombinált streamekből.

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Érték kibocsátása 1000ms-onként
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Érték kibocsátása 750ms-onként
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

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

Operátorok — zip

A zip megvárja, hogy érték érkezzen minden streamből, és ezekből az értékekből tömböt alkot. Ha egy streamből nem érkezik érték, a csoport nem fog létrejönni.

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Érték kibocsátása 1000ms-onként
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Érték kibocsátása 750ms-onként
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

const observer_3 = Observable.create((observer) => {
  let i = 1;
  // Érték kibocsátása 500ms-onként
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
});

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

Operátorok — forkJoin

A forkJoin szintén egyesíti a streameket, de csak akkor bocsát ki értéket, amikor az összes stream befejeződött (complete).

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Érték kibocsátása 1000ms-onként
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
}).pipe(take(3));

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Érték kibocsátása 750ms-onként
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
}).pipe(take(5));

const observer_3 = Observable.create((observer) => {
  let i = 1;
  // Érték kibocsátása 500ms-onként
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
}).pipe(take(4));

forkJoin(observer_1, observer_2, observer_3);

Operátorok — map

A map átalakító operátor módosítja a kibocsátott értéket egy újra.

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

Observable.create((observer) => {
  let i = 1;
  // Érték kibocsátása 1000ms-onként
  setInterval(() => {
    observer.next(i++);
  }, 1000);
}).pipe(
  map(x => x * 10),
  take(3)
);

Operátorok – share, tap

A tap operator allows you to perform side effects, that is, any actions that do not affect the sequence.

A utility operator share can turn a cold stream into a hot one.

We're done with operators. Let's move on to Subject.

Gondolatok hangosan

And I went to have some tea. These examples have tired me out 😀

Family of subjects

The family of subjects is a vivid example of hot streams. These classes are a kind of hybrid that serve as both observable and observer. Since a subject is a hot stream, you need to unsubscribe from it. Speaking of the main methods, they are:

next – passing new data into the stream
error – error and stream termination
complete – stream completion
subscribe – subscribe to the stream
unsubscribe – unsubscribe from the stream
asObservable – transform into an observer
toPromise – transforms into a promise

There are 4 to 5 types of subjects.

Gondolatok hangosan

I mentioned 4 on the stream, but it turned out they added another one. As they say, live and learn.

Simple Subject new Subject()– the simplest type of subjects. Created without parameters. Passes values that arrive only after subscription.

BehaviorSubject new BehaviorSubject( defaultData ) – in my opinion, the most common type of subjects. Takes a default value as input. Always keeps the last emitted data, which it sends upon subscription. This class also has a useful method value that returns the current value of the stream.

ReplaySubject new ReplaySubject(bufferSize?: number, windowTime?: number) — can optionally take the size of the buffer of values it will store as the first argument and the time during which we need the changes as the second.

AsyncSubject new AsyncSubject() — nothing happens upon subscription, and the value will only be returned upon complete. Only the last value of the stream will be returned.

WebSocketSubject new WebSocketSubject(urlConfigOrSource: string | WebSocketSubjectConfig | Observable, destination?: Observer) — The documentation is silent on this, and I see it for the first time myself. If anyone knows what it does, please write, and we will supplement this.

Phew. Well, we've covered everything I wanted to talk about today. I hope this information was useful. You can independently review the list of literature in the useful information tab.