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

Hej alla. Det är Omlnytsky Sergey här. För inte så länge sedan höll jag en ström om reaktiv programmering, där jag pratade om asynkronitet i JavaScript. Idag skulle jag vilja sammanfatta detta material.

Men innan vi börjar med huvuddelen av materialet behöver vi göra en introduktion. Så, låt oss börja med definitionerna: vad är en stack och en kö?
Stack — är en samling där elementen hämtas enligt principen "senast in, först ut" LIFO.
Kö — är en samling där elementen hämtas enligt principen "först in, först ut" FIFO.

Okej, vi fortsätter.

JavaScript är ett enskilt trådat programmeringsspråk. Det betyder att det bara finns en körtråd och en stack där funktionerna placeras i kö för utförande. Följaktligen kan JavaScript vid ett givet ögonblick bara utföra en operation, medan andra operationer väntar på sin tur i stacken tills de kallas.

Anropsstack — är en databasstruktur som, förenklat sagt, registrerar information om var vi befinner oss i programmet. När vi går in i en funktion, placerar vi en post om den högst upp i stacken. När vi återvänder från funktionen tar vi bort det översta elementet från stacken och hamnar där vi kallade funktionen. Det är allt som stacken kan göra. Men nu uppstår den ytterst intressanta frågan. Hur fungerar då asynkroniteten i JavaScript?

I själva verket finns det, förutom stacken, en speciell kö i webbläsarna för arbete med så kallat WebAPI. Funktioner från denna kö kommer att utföras i ordning först efter att stacken är helt rensad. Endast efter detta placeras de från kön i stacken för utförande. Om det för närvarande finns minst ett element i stacken, kan de inte komma in i stacken. Just av den anledningen är anrop av funktioner via timeout ofta tidsmässigt oexakta, eftersom funktionen inte kan komma från kön till stacken så länge den är fylld.

Låt oss titta på följande exempel och gå igenom dess "utförande" steg för steg. Låt oss också titta på vad som händer i systemet.

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

1) Inga händelser för tillfället. Webbläsarens konsol är tom, anropsstacken är tom.

2) Sedan läggs kommandot console.log('Hi') till anropsstacken.

3) Och den utförs

4) Sedan tas console.log('Hi') bort från anropsstacken.

5) Nu går vi vidare till kommandot setTimeout(function cb1() {… }). Det läggs till i anropsstacken.

6) Kommandot setTimeout(function cb1() {… }) utförs. Webbläsaren skapar en timer, som är en del av Web API. Den kommer att räkna ner tiden.

7) Kommandot setTimeout(function cb1() {… }) har slutfört sin uppgift och tas bort från anropsstacken.

8) Kommandot console.log('Bye') läggs till i anropsstacken.

9) Kommandot console.log('Bye') utförs.

10) Kommandot console.log('Bye') tas bort från anropsstacken.

11) När minst 5000 ms har gått, avslutar timern sin uppgift och placerar callback cb1 i callback-kön.

12) Event loop tar funktionen cb1 från callback-kön och lägger den i anropsstacken.

13) Funktionen cb1 utförs och lägger console.log('cb1') i anropsstacken.

14) Kommandot console.log('cb1') utförs.

15) Kommandot console.log('cb1') tas bort från anropsstacken.

16) Funktionen cb1 tas bort från anropsstacken.

Låt oss titta på exemplet i dynamik:

Nu har vi gått igenom hur asynkronitet är implementerat i JavaScript. Låt oss kort diskutera evolutionen av asynkron kod.

Evolutionen av asynkron kod.

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 känner det i JavaScript, kan endast genomföras av funktioner. De kan överföras som vilken annan variabel som helst till andra funktioner. Så föddes callbacks. Och det är roligt och underhållande, tills det förvandlas till sorg och elände. Varför? Jo, helt enkelt:

• Med den ökande komplexiteten av koden, förvandlas projektet snabbt till svårbegripliga djupt inbäddade block — "callback hell".
• Felhantering kan lätt förbises.
• Det går inte att returnera uttryck med return.

Med införandet av Promise blev läget något bättre.

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

• Det kom kedjor av promisser, vilket förbättrade kodens läsbarhet.
• Det kom en separat metod för att fånga fel.
• Möjligheten till parallell exekvering har dykt upp med hjälp av Promise.all
• Vi kan lösa inbäddad asynkronitet med hjälp av async/await

Men ett löfte har sina begränsningar. Till exempel kan ett löfte, utan extra krångel, inte avbrytas, och det viktigaste är - det fungerar med ett enda värde.

Nu har vi smidigt kommit till reaktiv programmering. Trött? Tja, du kan brygga en kopp te, fundera lite och återvända för att läsa vidare. Jag fortsätter.

Reaktiv programmeringär en programmeringsparadigm inriktad på datastreamar och spridning av förändringar. Låt oss granska vad en datastream verkligen är.

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

const eventsArray = [];

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

Tänk på att vi har ett inmatningsfält. Vi skapar en array, och på varje keyup-händelse från input kommer vi att lagra händelsen i vår array. Jag vill påpeka att vår array är sorterad efter tid - det vill säga, indexet för senare händelser är större än indexet för tidigare. En sådan array representerar en förenklad modell av en datastream, men det är fortfarande inte en datastream. För att denna array ska kunna kallas en datastream måste den kunna meddela prenumeranterna om att nya data har kommit in. På så sätt har vi kommit fram till definitionen av en datastream.

Datastream

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

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

Datastreamär en array av data sorterade efter tid som kan meddela när data har ändrats. Tänk nu på hur bekvämt det blir att skriva kod där en enda åtgärd kräver att flera händelser anropas på olika ställen i koden. Vi gör helt enkelt en prenumeration på datastreamen och den meddelar oss själv när förändringar inträffar. Detta kan biblioteket RxJs göra.

RxJS är ett bibliotek för att arbeta med asynkrona och händelsebaserade program med hjälp av observerbara sekvenser. Biblioteket tillhandahåller den grundläggande typen Observable, några hjälptyper (Observer, Schedulers, Subjects) och operatörer för att hantera händelser som samlingar (map, filter, reduce, every och liknande från JavaScript Array).

Låt oss gå igenom de grundläggande begreppen i detta bibliotek.

Observable, Observer, Producer

Observable — den första grundläggande typen som vi kommer att titta på. Denna klass innehåller huvuddelen av implementationen av RxJs. Den är kopplad till det observerade flödet som man kan prenumerera på med metoden subscribe.

I Observable implementeras en hjälpfunktion för att skapa uppdateringar, så kallad Observer. Källan till värden för Observer kallas Producer. Det kan vara en array, en iterator, web socket, något evenemang osv. Så man kan säga att observable är en förmedlare mellan Producer och Observer.

Observable hanterar tre typer av händelser för Observer:

• next – ny data
• error – ett fel om sekvensen avslutades på grund av ett undantag. Denna händelse innebär också att sekvensen avslutas.
• complete — en signal om att sekvensen är avslutad. Det betyder att det inte kommer att finnas fler data.

Låt oss se en demo:

I början kommer vi att bearbeta värdena 1, 2, 3, och efter 1 sekund kommer vi att få 4 och avsluta vårt flöde.

Tankar högt

Och här insåg jag att det var mer intressant att berätta än att skriva om det. 😀

Subscription

När vi prenumererar på flödet skapar vi en ny klass subscription, som ger oss möjlighet att avbryta prenumerationen med metoden unsubscribe. Vi kan också gruppera prenumerationer med metoden add. Och det är logiskt att vi kan avgruppera flöden med hjälp av ta bort. Metoderna add och remove tar en annan prenumeration som inparameter. Jag vill påpeka att när vi avbryter en prenumeration så avbryter vi också alla underordnade prenumerationer som om vi hade anropat metoden unsubscribe på dem. Låt oss gå vidare.

Typer av flöden

Om jag skulle föra en liknelse, skulle jag jämföra en varm ström med en film på bio. Från det ögonblick du anlände, började du titta. Den kalla strömmen skulle jag likna vid ett samtal till kundsupport. Den som ringer lyssnar på telefonsvararens meddelande från början till slut, men du kan lägga på luren genom att avprenumerera.

Jag vill påpeka att det också finns så kallade varma strömmar (detta begrepp har jag sett extremt sällan, och då endast på utländska forum) - det är en ström som transformeras från kall till varm. Frågan uppstår - var ska man använda det? Jag kommer att ge ett exempel från praktiken.

Jag arbetar med Angular. Det använder aktivt rxjs. För att hämta data från servern förväntar jag mig en kall ström och använder denna ström i mallen med hjälp av asyncPipe. Om jag använder denna pipe flera gånger, och återgår till definitionen av kall ström, kommer varje pipe att begära data från servern, vilket med andra ord är märkligt. Men om jag omvandlar den kalla strömmen till en varm, så kommer begäran att ske en gång.

Att förstå typer av strömmar är generellt ganska svårt för nybörjare, men det är viktigt.

Operators

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

Möjligheterna att arbeta med strömmar utvidgas av operatörer. De hjälper till att hantera händelser som passerar genom Observable. Vi kommer att titta på några av de mest populära, och mer detaljerad information om operatörer kan hittas via länkarna i den användbara informationen.

Operators — of

Vi börjar med hjälpoperatören of. Den skapar en Observable baserat på ett enkelt värde.

Operators — filter

Filteroperatören filter, som namnet antyder, filtrerar signalen i strömmen. Om operatören returnerar sant, släpps den igenom.

Operators — take

take - tar antalet emiterade värden som följande ström kommer att avsluta efter.

Operators — debounceTime

debounceTime - kastar bort de emitterade värdena som faller inom det angivna tidsintervallet mellan utsignalerna - efter det tidsintervall, emitteras det sista värdet.

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

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Skickar värdet var 1000 ms
  setInterval(() => {
    observer.next(i++)
  }, 1000);

  // Skickar värdet var 1500 ms
  setInterval(() => {
    observer.next(i++)
  }, 1500);
}).pipe(
  debounceTime(700),  // Väntar 700 ms med värden innan de bearbetas
  take(3)
);  

Operatörer — takeWhile

Avger värden så länge takeWhile inte returnerar false, efter det avregistreras från strömmen.

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

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Skickar värdet var 1000 ms
  setInterval(() => {
    observer.next(i++)
  }, 1000);
}).pipe(
  takeWhile(producer => producer < 5)
);  

Operatörer — combineLatest

CombineLatest-operatören liknar promise.all. Den slår samman flera strömmar till en. När varje ström har gjort minst en avvikelse får vi de senaste värdena från varje som en array. Efter vilket som helst avvikelse från de sammanfogade strömmarna kommer den att ge nya värden.

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Skickar värdet var 1000 ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Skickar värdet var 750 ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

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

Operatörer — zip

Zip — väntar på värde från varje ström och formar en array baserat på dessa värden. Om inget värde kommer från någon ström kommer gruppen inte att bildas.

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Skickar värdet var 1000 ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Skickar värdet var 750 ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

const observer_3 = Observable.create((observer) => {
  let i = 1;
  // Skickar värdet var 500 ms
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
});

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

Operatörer — forkJoin

forkJoin slår också samman strömmarna, men den avger värden endast när alla strömmar har avslutats (komplett).

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Sänder värdet var 1000ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
}).pipe(take(3));

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Sänder värdet var 750ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
}).pipe(take(5));

const observer_3 = Observable.create((observer) => {
  let i = 1;
  // Sänder värdet var 500ms
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
}).pipe(take(4));

forkJoin(observer_1, observer_2, observer_3);

Operatorer — map

Transformationsoperatorn map omvandlar värdet som emitteras till ett nytt.

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

Observable.create((observer) => {
  let i = 1;
  // Sänder värdet var 1000ms
  setInterval(() => {
    observer.next(i++);
  }, 1000);
}).pipe(
  map(x => x * 10),
  take(3)
);

Operatorer – share, tap

Operatorn tap tillåter sid-effekter, det vill säga åtgärder som inte påverkar sekvensen.

Hjälpoperatorn share kan göra en kallström till en varmström.

Vi är klara med operatorerna. Låt oss gå vidare till Subject.

Tankar högt

Och här gick jag för att dricka te. Dessa exempel tröttade ut mig 😀

Familjen subject-er

Familjen subject-er är ett tydligt exempel på varma strömmar. Dessa klasser är en slags hybrid som fungerar både som observable och observer. Eftersom subject är en varmström, måste man avregistrera sig från den. De huvudsakliga metoderna är:

• next – skicka nya data till strömmen
• error – fel och avslutning av strömmen
• complete – avslutning av strömmen
• subscribe – prenumerera på strömmen
• unsubscribe – avregistrera sig från strömmen
• asObservable – transformera till en observer
• toPromise – omvandlar till ett löfte

Det finns 4–5 typer av subject-er.

Tankar högt

På streamen pratade jag om 4, men sedan visade det sig att de hade lagt till en till. Som man säger, lär sig något nytt varje dag.

Enkel Subject new Subject()– den mest grundläggande typen av subject-er. Skapas utan parametrar. Skickar värden som kommer efter registrering.

BehaviorSubject new BehaviorSubject(defaultData) – enligt mig den mest spridda typen av subject-er. Tar ett standardvärde som argument. Sparar alltid det senaste emiterade värdet som den skickar vid registrering. Denna klass har också en användbar metod, value, som returnerar det aktuella värdet i strömmen.

ReplaySubject new ReplaySubject(bufferSize?: number, windowTime?: number) – kan som första argument valfritt ta storleken på buffern av värden som den kommer att lagra och som andra argument tiden under vilken vi behöver ändringar.

AsyncSubject ny AsyncSubject() — vid prenumeration händer ingenting, och värdet kommer att returneras endast vid complete. Endast det senaste värdet från strömmen kommer att returneras.

WebSocketSubject ny WebSocketSubject(urlConfigOrSource: string | WebSocketSubjectConfig<T> | Observable<T>, destination?: Observer<T>) — Dokumentationen nämner inte detta och jag ser det för första gången själv. Den som vet vad det gör, skriv gärna så vi kan komplettera.

Puh. Nåväl, vi har nu gått igenom allt jag ville berätta idag. Jag hoppas att denna information har varit nyttig. Du kan själv bekanta dig med litteraturlistan under fliken nyttig information.

Nyttig information

• https://vk.com/frontend_practice?w=wall-182297870_931 — länk till streamen
• https://habr.com/ru/company/ruvds/blog/337042/ — Hur JS fungerar: översikt av motorn, körrutiner, samt anropsstackar
• https://habr.com/ru/company/ruvds/blog/340508 — Hur JS fungerar: evenemangslopp, asynkronitet och fem sätt att förbättra koden med hjälp av async / await
• http://vanilla-js.blogspot.com/2014/09/event-loop-javasript.html — Hur Event Loop fungerar i JavaScript
• https://tproger.ru/translations/asynchronous-javascript/ — Utvecklingen av asynkront JavaScript
• https://habr.com/ru/company/ruvds/blog/341880/ — Vad är RxJS och varför är det bra att känna till det
• https://medium.com/ngx/practical-use-rxjs-81aaab57045c — Praktisk tillämpning av RxJS
• https://coursetro.com/posts/code/148/RxJS-Observables-Tutorial—Creating-&-Subscribing-to-Observables — RxJS Observables Tutorial — Skapa & Prenumerera på Observables
• https://slides.com/zenabsaleem/rxjs-hot-cold-observables#/ — RXJS: Varma och kalla Observables
• https://clck.ru/FiPgB — Klasser, funktioner för att skapa Observable. Operatörer.
• https://www.learnrxjs.io/operators/ — RxJS-operatörer med exempel
• https://rxjs.dev/api — API-lista
• https://clck.ru/FkS8Y — Olika slags Subject och Scheman i RxJS

Källa: habr.com