🥇Asynkron programmering i JavaScript. (Callback, Promise, RxJs)

Hej alle sammen. Det er Omelnitsky Sergey her. For nylig havde jeg en stream om reaktiv programmering, hvor jeg talte om asynkronitet i JavaScript. I dag vil jeg gerne opsummere dette materiale.

Men før vi går i gang med hovedmaterialet, skal vi lave en introduktion. Så lad os starte med definitioner: hvad er en stak og en kø?
Stak — er en samling, hvis elementer bliver hentet efter princippet 'sidst ind, først ud' LIFO
Kø — er en samling, hvis elementer bliver hentet efter princippet ('først ind, først ud' FIFO

Okay, lad os fortsætte.

JavaScript er et enkelttrådet programmeringssprog. Det betyder, at der kun er én udførelsestråd og én stak, hvor funktioner bliver sat i kø til udførelse. Derfor kan JavaScript kun udføre én operation ad gangen, mens de andre operationer venter i stakken, indtil de bliver kaldt.

Stak af kald — er en dataramme, der forenklet sagt gemmer oplysninger om, hvor vi befinder os i programmet. Hvis vi går ind i en funktion, placerer vi en post om den øverst i stakken. Når vi vender tilbage fra funktionen, fjerner vi det øverste element fra stakken og befinder os der, hvor vi kaldte denne funktion fra. Det er alt, hvad stakken kan.

Og nu et meget interessant spørgsmål. Hvordan fungerer asynkronitet i JavaScript så?

Faktisk, udover stakken, eksisterer der i browsere en særlig kø til arbejde med såkaldte WebAPI'er. Funktioner fra denne kø udføres i rækkefølge, først efter at stakken er helt tømt. Først derefter placeres de fra køen til stakken til udførelse. Hvis der i stakken i øjeblikket er mindst ét element, kan de ikke komme ind i stakken. Det er netop derfor, funktioner med timeout ofte ikke er tidsmæssigt præcise, da funktionen ikke kan komme fra køen til stakken, mens den er fyldt.

Lad os overveje følgende eksempel og tage os af dets trin-for-trin 'udførelse'. Vi vil også se, hvad der sker i systemet.

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

1) Indtil videre sker der ikke noget. Browsers konsol er ren, stakken af kald er tom.

3) Og den udføres

4) Derefter fjernes console.log(‘Hi’) fra call-stakken.

5) Nu går vi videre til kommandoen setTimeout(function cb1() {… }). Den tilføjes til call-stakken.

6) Kommandoen setTimeout(function cb1() {… }) udføres. Browseren opretter en timer, som er en del af Web API. Den vil udføre nedtællingen.

7) Kommandoen setTimeout(function cb1() {… }) er færdig og fjernes fra call-stakken.

8) Kommandoen console.log(‘Bye’) tilføjes til call-stakken.

9) Kommandoen console.log(‘Bye’) udføres.

10) Kommandoen console.log(‘Bye’) fjernes fra call-stakken.

11) Efter mindst 5000 ms er gået, afslutter timeren sit arbejde og placerer callback cb1 i callback-køen.

12) Event-loopet tager funktionen cb1 fra callback-køen og placerer den i call-stakken.

13) Funktionen cb1 udføres og tilføjer console.log(‘cb1’) til call-stakken.

14) Kommandoen console.log(‘cb1’) udføres.

15) Kommandoen console.log(‘cb1’) fjernes fra call-stakken.

16) Funktionen cb1 fjernes fra call-stakken.

Lad os se på et eksempel i dynamik:

Sådan, nu har vi set, hvordan asynkronitet er implementeret i JavaScript. Lad os kort diskutere udviklingen af asynkron kode.

Udviklingen af asynkron kode.

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

Asynkront programmering, som vi kender det i JavaScript, kan kun implementeres ved hjælp af funktioner. De kan gives som enhver anden variabel til andre funktioner. Sådan blev callbacks til. Og det er sjovt, livligt og underholdende, indtil det bliver trist, melankolsk og sørgeligt. Hvorfor? Jo, alt er enkelt:

Med stigende kompleksitet i koden bliver projektet hurtigt til svære at forstå dybt indlejrede blokke — 'callback hell'.
Fejlbehandling kan nemt blive overset.
Du kan ikke returnere udtryk med return.

Med fremkomsten af Promises blev situationen lidt bedre.

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

Der opstod kæder af promises, hvilket forbedrede læsbarheden af koden
Der kom en separat metode til at opfange fejl
Der er nu mulighed for parallelt udførsel ved hjælp af Promise.all
Vi kan løse indlejret asynkronitet ved hjælp af async/await

Men et løfte har sine begrænsninger. For eksempel kan et løfte ikke annulleres uden nogle tricks, og vigtigst af alt - det arbejder kun med én værdi.

Nå, vi er stille og roligt ved at nærme os reaktiv programmering. Er du træt? Men heldigvis kan du tage en kop te, tænke over det og komme tilbage for at læse videre. Jeg fortsætter.

Reaktiv programmering — en programmeringsparadigme, der fokuserer på datastreams og ændringsformidling. Lad os se nærmere på, hvad en datastream er.

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

const eventsArray = [];

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

Forestil dig, at vi har et inputfelt. Vi opretter et array, og ved hver keyup-hændelse fra inputfeltet vil vi gemme hændelsen i vores array. Det skal bemærkes, at vores array er sorteret efter tid, dvs. indekset for senere hændelser er større end indekset for tidligere hændelser. Et sådant array repræsenterer en forenklet model af en datastream, men det er endnu ikke en stroom. For at dette array sikkert kan kaldes en stream, skal det kunne kommunikere til abonnenterne, når nye data ankommer. Så vi er kommet til definitionen af en stream.

Datastream

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

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

Stream — er et array af data, der er sorteret efter tid, som kan informere om, at dataene er ændret. Forestil dig, hvor praktisk det bliver at skrive kode, hvor en handling kræver, at flere hændelser kaldes på forskellige steder i koden. Vi abonnerer blot på strømmen, og den vil selv informere os om, hvornår der sker ændringer. Dette kan RxJs bibliotekt gøre.

RxJS — er et bibliotek til at arbejde med asynkrone og event-baserede programmer ved hjælp af observerbare sekvenser. Biblioteket tilbyder den grundlæggende type Observable, flere hjælpefunktioner (Observer, Schedulers, Subjects) og operatører til at arbejde med hændelser som samlinger (map, filter, reduce, every og lignende fra JavaScript Array).

Lad os se nærmere på de grundlæggende begreber i dette bibliotek.

Observable, Observer, Producer

Observable — den første grundlæggende type, som vi vil se på. Denne klasse indeholder den vigtigste del af RxJs-implementeringen. Den er knyttet til en observerbar stream, som man kan abonnere på ved hjælp af metoden subscribe.

I Observable implementerer en støttefunktion til at skabe opdateringer, såkaldt Observer. Kilden til værdierne for Observer kaldes Producer. Dette kan være et array, en iterator, en web socket, en begivenhed osv. Så man kan sige, at observable fungerer som en formidler mellem Producer og Observer.

Observable håndterer tre typer begivenheder for Observer:

next – nye data
error – en fejl, hvis sekvensen afsluttes på grund af en undtagelsessituation. Denne begivenhed betyder også, at sekvensen er afsluttet.
complete – et signal om, at sekvensen er afsluttet. Det betyder, at der ikke vil komme flere nye data.

Lad os se en demo:

I starten vil vi behandle værdierne 1, 2, 3, og efter 1 sekund vil vi modtage 4 og afslutte vores strøm.

Tankerne højt

Og her indså jeg, at det var mere interessant at fortælle end at skrive om det. 😀

Subscription

Når vi tilmelder os en strøm, opretter vi en ny klasse subscription, som giver os mulighed for at afmelde os med metoden unsubscribe. Vi kan også gruppere abonnementer ved hjælp af metoden add. Samt, det er logisk, at vi kan afgruppere strømme ved hjælp af remove. Metoderne add og remove tager en anden abonnement som input. Det skal bemærkes, at når vi afmelder os, afmelder vi os alle underordnede abonnementer, som om vi også havde kaldt metoden unsubscribe på dem. Lad os fortsætte.

Typer af strømme

HOT	COLD
Producer oprettes uden for observable	Producer oprettes inden for observable
Data overføres på tidspunktet for oprettelsen af observable	Data gives under tilmelding
Der er behov for ekstra logik for at afmelde	Strømmen afsluttes af sig selv
Bruger en en-til-mange forbindelse	Bruger en en-til-en forbindelse
Alle abonnementer har en fælles værdi	Abonnementer er uafhængige
Data kan gå tabt, hvis der ikke er nogen tilmelding	Genudgiver alle værdier fra strømmen til et nyt abonnement

Hvis jeg skulle lave en analogi, ville jeg beskrive en varm strøm som en film i biografen. Uanset hvornår du ankom, begynder du at se filmen fra det øjeblik. En kold strøm ville jeg sammenligne med et opkald til teknisk support. Enhver, der ringer, lytter til auto-svareren fra start til slut, men du kan lægge røret på ved hjælp af unsubscribe.

Jeg vil gerne bemærke, at der også findes såkaldte varme strømme (dette udtryk har jeg kun set sjældent og kun i udenlandske fællesskaber) — det er en strøm, der omdannes fra en kold strøm til en varm. Spørgsmålet opstår — hvor skal det bruges)) Jeg vil give et eksempel fra praksis.

Jeg arbejder med Angular. Den bruger aktivt RxJS. For at hente data fra serveren forventer jeg en kold strøm, og jeg bruger denne strøm i skabelonen med asyncPipe. Hvis jeg bruger denne pipe flere gange, så, tilbage til definitionen af en kold strøm, vil hver pipe anmode om data fra serveren, hvilket mildt sagt er underligt. Men hvis jeg omdanner den kolde strøm til en varm, vil anmodningen ske én gang.

Generelt set er forståelsen af strømmens typer ret vanskelig for begyndere, men den er vigtig.

Operatører

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

Operatører udvider vores mulighed for at arbejde med strømme. De hjælper med at kontrollere de hændelser, der forekommer i Observable. Vi vil se på et par af de mest populære, og mere detaljeret om operatørerne kan findes via links i nyttig information.

Operatører — of

Lad os starte med hjælpeoperatøren of. Den skaber en Observable baseret på en simpel værdi.

Operatører — filter

Filteroperatøren filter, som man kan forstå ud fra navnet, filtrerer signalet i strømmen. Hvis operatøren returnerer sandhed, så sendes det videre.

Operatører — take

take — Modtager antallet af emissioner, hvorefter strømmen afsluttes.

Operatører — debounceTime

debounceTime — kasserer de udsendte værdier, der falder inden for den angivne tidsramme mellem udgangsværdierne — efter et tidsinterval udsendes den sidste værdi.

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

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Udsender værdi hvert 1000 ms
  setInterval(() => {
    observer.next(i++)
  }, 1000);

  // Udsender værdi hver 1500 ms
  setInterval(() => {
    observer.next(i++)
  }, 1500);
}).pipe(
  debounceTime(700),  // Venter 700 ms på værdierne før behandling
  take(3)
);

Operatører — takeWhile

Udsender værdier, så længe takeWhile ikke returnerer false, hvorefter den afmelder sig fra strømmen.

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

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Udsender værdi hvert 1000ms
  setInterval(() => {
    observer.next(i++)
  }, 1000);
}).pipe(
  takeWhile(producer => producer < 5)
);

Operatører — combineLatest

Den kombinerede operator combineLatest ligner promise.all. Den kombinerer flere strømme til én. Når hver strøm har sendt mindst én værdi, modtager vi de seneste værdier fra hver strøm i en array. Derefter, efter enhver emit fra de kombinerede strømme, vil den returnere nye værdier.

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Udsender værdi hvert 1000ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Udsender værdi hvert 750ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

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

Operatører — zip

Zip venter på en værdi fra hver strøm og danner en array baseret på disse værdier. Hvis der ikke kommer en værdi fra en af strømmen, vil gruppen ikke blive dannet.

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Udsender værdi hvert 1000ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Udsender værdi hvert 750ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

const observer_3 = Observable.create((observer) => {
  let i = 1;
  // Udsender værdi hvert 500ms
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
});

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

Operatører — forkJoin

forkJoin kombinerer også strømme, men den udsender værdi kun når alle strømme er afsluttet (complete).

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Udsender værdi hvert 1000ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
}).pipe(take(3));

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Udsender værdi hvert 750ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
}).pipe(take(5));

const observer_3 = Observable.create((observer) => {
  let i = 1;
  // Udsender værdi hvert 500ms
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
}).pipe(take(4));

forkJoin(observer_1, observer_2, observer_3);

Operatører — map

Transformationsoperatoren map konverterer den udsendte værdi til en ny.

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

Observable.create((observer) => {
  let i = 1;
  // Udsender værdi hvert 1000ms
  setInterval(() => {
    observer.next(i++);
  }, 1000);
}).pipe(
  map(x => x * 10),
  take(3)
);

Operatører – share, tap

Tap-operatoren tillader sideeffekter, det vil sige handlinger, der ikke påvirker sekvensen.

Utility-operatoren share kan gøre en kold strøm varm.

Vi er færdige med operatorerne. Lad os gå videre til Subject.

Tankerne højt

Og så gik jeg for at drikke noget te. Disse eksempler har trættet mig 😀

Familien af subjects

Familien af subjects er et klart eksempel på varme strømme. Disse klasser er en form for hybrid, der fungerer som både observable og observer. Da subject er en varm strøm, skal man tilmelde sig det. De grundlæggende metoder er:

next – sender nye data til strømmen
error – fejl og afslutning af strømmen
complete – afslutning af strømmen
subscribe – tilmeld dig strømmen
unsubscribe – afmeld dig strømmen
asObservable – transformer til observer
toPromise – transformer til et løfte

Der skelnes mellem 4 til 5 typer subjects.

Tankerne højt

Jeg sagde 4 på streamen, men det viste sig, at de havde tilføjet endnu en. Som man siger, lever og lær.

Simpel Subject new Subject()– den enkleste type af subjects. Oprettes uden parametre. Sender værdier, der kun kommer efter tilmelding.

BehaviorSubject new BehaviorSubject(defaultData) – efter min mening den mest udbredte type of subjects. Tager en standardværdi som input. Bevarer altid data fra den seneste emission, som den sender ved tilmelding. Denne klasse har også en nyttig metode value, som returnerer den nuværende værdi i strømmen.

ReplaySubject new ReplaySubject(bufferSize?: number, windowTime?: number) — Kan tage størrelsen af bufferet som første argument og tiden, hvor vi har brug for ændringer, som det andet.

AsyncSubject new AsyncSubject() — ved tilmelding sker der ikke noget, og værdien returneres først ved complete. Kun den seneste værdi fra strømmen vil blive returneret.

WebSocketSubject new WebSocketSubject(urlConfigOrSource: string | WebSocketSubjectConfig | Observable, destination?: Observer) — om det er dokumentationen stille, og jeg ser det for første gang selv. Hvis nogen ved, hvad det gør, skriv, så tilføjer vi det.

Puh. Så har vi dækket det hele, jeg ønskede at tale om i dag. Jeg håber, denne information har været nyttig. Du kan selv se litteraturlisten under fanen nyttig information.