🥇Asynchroon programmeren in JavaScript. ( Callback, Promise, RxJs )

Hallo iedereen. Sergei Omelnitsky hier. Onlangs heb ik een livestream gedaan over reactief programmeren, waarin ik sprak over asynchroniciteit in JavaScript. Vandaag wil ik dit materiaal samenvatten.

Maar voordat we met het hoofdonderwerp beginnen, moeten we een inleiding maken. Laten we beginnen met definities: wat zijn een stapel en een wachtrij?
Stapel is een collectie waarvan de elementen volgens het principe 'laatst binnen, eerst eruit' LIFO worden verkregen.
Wachtrij is een collectie waarvan de elementen volgens het principe 'eerste in, eerste uit' FIFO worden verkregen.

Oké, laten we verder gaan.

JavaScript is een single-threaded programmeertaal. Dit betekent dat er slechts één uitvoeringsdraad en één stapel is waarin functies in de wachtrij voor uitvoering worden geplaatst. Dus kan JavaScript op elk moment slechts één operatie uitvoeren; andere operaties moeten wachten in de stapel totdat ze worden opgeroepen.

De oproepstapel is een datastructuur die samengevat informatie opschrijft over waar we in het programma zijn. Wanneer we een functie ingaan, plaatsen we een record daarvan bovenaan de stapel. Wanneer we uit de functie terugkeren, halen we het bovenste element van de stapel en komen we daar waar we deze functie hebben opgeroepen. Dat is alles wat de stapel doet. En nu een zeer interessante vraag. Hoe werkt asynchroniciteit in JavaScript dan?

In feite is er naast de stapel in browsers een speciale wachtrij voor het werken met het zogenaamde WebAPI. Functies uit deze wachtrij worden alleen in volgorde uitgevoerd nadat de stapel volledig is geleegd. Pas daarna worden ze van de wachtrij naar de stapel verplaatst voor uitvoering. Als er momenteel minstens één element in de stapel zit, kunnen ze de stapel niet binnenkomen. Dit is precies de reden waarom aanroeptijden van functies met time-outs vaak onnauwkeurig zijn, omdat een functie de stapel niet binnen kan gaan zolang deze vol is.

Laten we het volgende voorbeeld bekijken en het stap voor stap 'uitvoeren'. Laten we ook zien wat er in het systeem gebeurt.

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

1) Ondertussen gebeurt er niets. De browserconsole is leeg, de oproepstapel is leeg.

2) Vervolgens wordt het commando console.log('Hi') aan de oproepstapel toegevoegd.

3) En het wordt uitgevoerd.

4) Vervolgens wordt console.log('Hi') uit de oproepstapel verwijderd.

5) Laten we nu kijken naar het commando setTimeout(function cb1() {… }). Het wordt aan de call stack toegevoegd.

6) Het commando setTimeout(function cb1() {… }) wordt uitgevoerd. De browser creëert een timer die onderdeel is van de Web API. Deze zal de tijd aftellen.

7) Het commando setTimeout(function cb1() {… }) is voltooid en wordt uit de call stack verwijderd.

8) Het commando console.log(‘Bye’) wordt aan de call stack toegevoegd.

9) Het commando console.log(‘Bye’) wordt uitgevoerd.

10) Het commando console.log(‘Bye’) wordt uit de call stack verwijderd.

11) Nadat er minstens 5000 ms zijn verstreken, voltooit de timer zijn werk en plaatst de callback cb1 in de callback queue.

12) De event loop neemt de functie cb1 uit de callback queue en plaatst deze in de call stack.

13) De functie cb1 wordt uitgevoerd en voegt console.log(‘cb1’) aan de call stack toe.

14) Het commando console.log(‘cb1’) wordt uitgevoerd.

15) Het commando console.log(‘cb1’) wordt uit de call stack verwijderd.

16) De functie cb1 wordt uit de call stack verwijderd.

Laten we eens kijken naar een voorbeeld in actie:

Dus we hebben besproken hoe asynchroniciteit in JavaScript wordt gerealiseerd. Laten we nu kort praten over de evolutie van asynchrone code.

De evolutie van asynchrone code.

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

Asynchroon programmeren, zoals we het in JavaScript kennen, kan enkel worden gerealiseerd met functies. Ze kunnen zoals elke andere variabele aan andere functies worden doorgegeven. Zo zijn callbacks ontstaan. En dat is leuk, grappig en spannend, totdat het verandert in verdriet, melancholie en verdriet. Waarom? Gewoon omdat:

Met de toenemende complexiteit van de code verandert het project al snel in moeilijk te begrijpen, meerdere geneste blokken — 'callback hell'.
Foutafhandeling kan gemakkelijk worden gemist.
Je kunt geen waarden teruggeven met return.

Met de komst van Promise is de situatie iets beter geworden.

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

Er zijn ketens van beloftes ontstaan, wat de leesbaarheid van de code verbeterde.
Er is een aparte methode voor foutafhandeling gekomen.
Er is de mogelijkheid voor parallelle uitvoering met Promise.all gekomen.
We kunnen geneste asynchroniteit oplossen met async/await

Maar een belofte heeft zijn eigen beperkingen. Bijvoorbeeld, een belofte kan niet worden geannuleerd zonder ingewikkelde trucs, en het belangrijkste is dat het met één waarde werkt.

Hier zijn we dan bij reactief programmeren. Moe? Gelukkig kun je even een kopje thee zetten, erover nadenken en later verder lezen. Ik ga intussen door.

Reactief programmeren — is een programmeerparadigma dat zich richt op datastromen en het verspreiden van wijzigingen. Laten we eens dieper ingaan op wat een datastroom is.

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

const eventsArray = [];

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

Stel je voor dat we een invoerveld hebben. We creëren een array, en bij elke keyup gebeurtenis van de input slaan we het evenement op in onze array. Daarbij willen we opmerken dat onze array op tijd is gesorteerd, dat wil zeggen: de index van latere gebeurtenissen is groter dan die van eerdere. Zo'n array stelt een vereenvoudigd model van een datastroom voor, maar dit is nog geen stroom. Om deze array met recht een stroom te noemen, moet hij op de een of andere manier kunnen communiceren met abonnees dat er nieuwe gegevens zijn binnengekomen. Zo zijn we tot de definitie van een datastroom gekomen.

Datastroom

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

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

Stroom — is een array van gegevens, gesorteerd op tijd, die kan aangeven dat de gegevens zijn gewijzigd. En stel je voor hoe handig het wordt om code te schrijven waarbij één actie meerdere gebeurtenissen in verschillende delen van de code vereist. We abonneren ons gewoon op de stroom en deze zal ons zelf informeren wanneer er wijzigingen plaatsvinden. Dit kan de RxJs bibliotheek.

RxJS — is een bibliotheek voor het werken met asynchrone en gebeurtenisgestuurde programma's met behulp van observeerbare reeksen. De bibliotheek biedt het belangrijkste type Observable, enkele hulpprogramma's (Observer, Schedulers, Subjects) en operators voor het werken met gebeurtenissen als collecties (map, filter, reduce, every en soortgelijke uit de JavaScript Array).

Laten we de belangrijkste concepten van deze bibliotheek eens bekijken.

Observable, Observer, Producer

Observable is het eerste basisgegevenstype dat we zullen bespreken. Deze klasse bevat het belangrijkste deel van de implementatie van RxJs. Het is verbonden met de observeerbare stroom waarop je je kunt abonneren met de subscribe-methode.

In Observable is an auxiliary mechanism implemented for creating updates, known as Observer. The source of values for an Observer is called Producer. This can be an array, an iterator, a web socket, some event, etc. So we can say that an observable is a conduit between Producer and Observer.

Observable processes three types of Observer events:

next – new data
error – an error if the sequence is completed due to an exceptional situation. This event also implies the completion of the sequence.
complete – a signal that the sequence is finished. This means that there will be no more new data

Let's look at a demo:

At first, we will process the values 1, 2, 3, and after 1 second, we will get 4 and complete our stream.

Thinking out loud

And then I realized that telling the story was more interesting than writing about it. 😀

Subscription

When we subscribe to a stream, we create a new class subscription, which allows us to cancel the subscription using the method unsubscribe. We can also group subscriptions using the method add. And logically, we can ungroup streams using verwijder. The add and remove methods accept another subscription as input. I would like to note that when we unsubscribe, we unsubscribe from all child subscriptions as if we called the unsubscribe method on them. Let's move on.

Types of streams

HOT	COLD
Producer is created outside of observable	Producer is created inside observable
Data is transmitted at the moment of observable creation	Data is communicated at the moment of subscription
Additional logic is needed for unsubscription	The stream ends by itself
Uses a one-to-many relationship	Uses a one-to-one relationship
All subscriptions have a unified value	Subscriptions are independent
Data can be lost if there is no subscription	Reissues all values of the stream for the new subscription

If I were to draw an analogy, I would picture a hot stream as a movie in a cinema. At whatever moment you arrive, you start watching from that point. I would compare a cold stream to a call to tech support. Any caller listens to the auto-responder recording from start to finish, but you can hang up using unsubscribe.

Ik wil opmerken dat er ook zogenaamde warme stromen bestaan (dit begrip kwam ik uiterst zelden tegen en alleen in buitenlandse gemeenschappen) — dit is een stroom die transformeert van een koude stroom naar een hete. De vraag rijst — waar gebruik je dit voor)) Ik zal een voorbeeld uit de praktijk geven.

Ik werk met Angular. Het maakt actief gebruik van RxJS. Voor het ophalen van gegevens van de server verwacht ik een koude stroom en gebruik deze stroom in de sjabloon met behulp van AsyncPipe. Als ik deze pipe meerdere keren gebruik, dan zal, terugkerend naar de definitie van een koude stroom, elke pipe gegevens van de server opvragen, wat zacht gezegd vreemd is. Maar als ik de koude stroom omzet in een warme, zal de aanvraag slechts één keer plaatsvinden.

Eigenlijk is het begrip van de soorten stromen vrij complex voor beginners, maar het is belangrijk.

Operators

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

De mogelijkheid om met stromen te werken wordt ons geboden door operators. Ze helpen evenementen te beheersen die plaatsvinden binnen de Observable. We zullen een paar van de meest populaire bekijken, en voor een gedetailleerde uitleg over operators kun je de links in de nuttige informatie raadplegen.

Operators — of

Laten we beginnen met de hulpopperator 'of'. Hij creëert een Observable op basis van een eenvoudige waarde.

Operators — filter

De filteroperator filter, zoals je uit de naam kunt afleiden, filtert het stroomsignaal. Als de operator waar teruggeeft, laat hij het verder doorgaan.

Operators — take

take — Neemt het aantal emissies aan, waarna de stroom wordt beëindigd.

Operators — debounceTime

debounceTime — Negeert de uitgespuwde waarden die binnen het opgegeven tijdsinterval tussen de output komen — na het verstrijken van het tijdsinterval geeft het de laatste waarde uit.

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

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Geeft waarde elke 1000ms uit
  setInterval(() => {
    observer.next(i++)
  }, 1000);

 // Geeft waarde elke 1500ms uit
  setInterval(() => {
    observer.next(i++)
  }, 1500);
}).pipe(
  debounceTime(700),  // Wacht 700ms op waarde voordat deze verwerkt wordt
  take(3)
);

Operators — takeWhile

Geeft waarden uit zolang takeWhile geen false retourneert, waarna het zich afmeldt van de stroom.

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

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Stoot een waarde uit om de 1000 ms
  setInterval(() => {
    observer.next(i++)
  }, 1000);
}).pipe(
  takeWhile( producer =>  producer < 5 )
);

Operators — combineLatest

De gecombineerde operator combineLatest lijkt op promise.all. Het combineert meerdere streams tot één. Zodra elke stream ten minste één emit heeft gedaan, krijgen we de laatste waarden van elk in de vorm van een array. Vervolgens, na elke emissie van de gecombineerde streams, geeft het nieuwe waarden terug.

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Stoot een waarde uit om de 1000 ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Stoot een waarde uit om de 750 ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

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

Operators — zip

Zip wacht op een waarde uit elke stream en vormt een array op basis van deze waarden. Als er geen waarde uit een bepaalde stream komt, wordt de groep niet gevormd.

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Stoot een waarde uit om de 1000 ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Stoot een waarde uit om de 750 ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

const observer_3 = Observable.create((observer) => {
  let i = 1;
  // Stoot een waarde uit om de 500 ms
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
});

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

Operators — forkJoin

forkJoin combineert ook streams, maar het emit alleen een waarde wanneer alle streams zijn beëindigd (complete).

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Stoot een waarde uit om de 1000 ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
}).pipe(take(3));

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Stoot een waarde uit om de 750
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
}).pipe(take(5));

const observer_3 = Observable.create((observer) => {
  let i = 1;
  // Stoot een waarde uit om de 500
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
}).pipe(take(4));

forkJoin(observer_1, observer_2, observer_3);

Operators — map

De transformaties operator map verandert de waarde van de emit in een nieuwe.

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

Observable.create((observer) => {
  let i = 1;
  // Stoot een waarde uit om de 1000 ms
  setInterval(() => {
    observer.next(i++);
  }, 1000);
}).pipe(
  map(x => x * 10),
  take(3)
);

Operators – share, tap

De tap-operator maakt het mogelijk om side-effecten te creëren, dat wil zeggen, daden die de volgorde niet beïnvloeden.

De utility-operator share kan een koude stroom omzetten in een warme.

We hebben de operators afgerond. Laten we doorgaan naar Subject.

Thinking out loud

En toen ging ik thee drinken. Deze voorbeelden hebben me uitgeput 😀

Familie van subjects

De familie van subjects is een mooi voorbeeld van warme stromen. Deze klassen zijn een soort hybride die tegelijkertijd als observable en observer optreden. Omdat het subject een warme stroom is, moet je je afmelden. Wat betreft de belangrijkste methoden zijn dat:

next – nieuwe data doorgeven naar de stroom
error – fout en beëindiging van de stroom
complete – beëindiging van de stroom
subscribe – abonneren op de stroom
unsubscribe – afmelden van de stroom
asObservable – omzetten naar observer
toPromise – omzetten naar promise

Er zijn 4 tot 5 typen subjects.

Thinking out loud

Ik zei 4 op de stream, maar blijkbaar hebben ze er nog een toegevoegd. Zoals het gezegde luidt: 'een leven lang leren'.

Eenvoudig Subject new Subject()– de eenvoudigste vorm van subjects. Wordt zonder parameters gemaakt. Geeft alleen waarden door die na de aanmelding zijn gekomen.

BehaviorSubject new BehaviorSubject(defaultData) – naar mijn mening de meest voorkomende vorm van subjects. Neemt een standaardwaarde als invoer. Bewaart altijd de gegevens van de laatste emissie, die het doorgeeft bij aanmelding. Deze klasse heeft ook een nuttige waarde methode die de huidige waarde van de stroom retourneert.

ReplaySubject new ReplaySubject(bufferSize?: number, windowTime?: number) – Kan optioneel een buffermaat voor de waarden aannemen die het zal opslaan als eerste argument, en als tweede argument de tijd waarin we wijzigingen nodig hebben.

AsyncSubject new AsyncSubject() – bij aanmelding gebeurt er niets, en de waarde wordt alleen geretourneerd bij complete. Er zal alleen de laatste waarde van de stroom worden geretourneerd.

WebSocketSubject new WebSocketSubject(urlConfigOrSource: string | WebSocketSubjectConfig | Observable, destination?: Observer) – De documentatie zegt er niets over en ik zie het voor het eerst. Wie weet wat het doet, laat het me weten, dan vullen we het aan.

Phew. Nou, dat was alles wat ik vandaag wilde bespreken. Ik hoop dat deze informatie nuttig was. Je kunt de literatuurlijst zelf bekijken in het tabblad nuttige informatie.