🥇Asynchrone Programmierung in JavaScript. (Callback, Promise, RxJs)

Hallo zusammen. Hier ist Sergey Omelnitsky. Kürzlich habe ich einen Stream über reaktive Programmierung gehalten, in dem ich über Asynchronität in JavaScript sprach. Heute möchte ich diese Inhalte zusammenfassen.

Bevor wir mit dem Hauptmaterial beginnen, müssen wir eine Einführung geben. Lassen Sie uns also mit den Definitionen beginnen: Was sind Stack und Queue?
Stack ist eine Sammlung, deren Elemente nach dem Prinzip 'Last In, First Out' (LIFO) abgerufen werden.
Queue ist eine Sammlung, deren Elemente nach dem Prinzip 'First In, First Out' (FIFO) abgerufen werden.

In Ordnung, fahren wir fort.

JavaScript ist eine einsträngige Programmiersprache. Das bedeutet, dass es nur einen Ausführungsstrang und einen Stack gibt, in den Funktionen zur Ausführung eingereiht werden. Das heißt, zur gleichen Zeit kann JavaScript nur eine Operation ausführen, während andere Operationen in der Warteschlange im Stack warten, bis sie aufgerufen werden.

Call Stack — ist eine Datenstruktur, die vereinfacht gesagt Informationen über den Ort im Programm speichert, an dem wir uns befinden. Wenn wir in eine Funktion wechseln, fügen wir einen Eintrag über diese Funktion oben auf dem Stack hinzu. Wenn wir aus der Funktion zurückkehren, ziehen wir das oberste Element vom Stack und landen dort, wo wir diese Funktion aufgerufen haben. Das ist alles, was der Stack kann. Aber jetzt stellt sich eine äußerst interessante Frage: Wie funktioniert die Asynchronität in JavaScript?

Tatsächlich gibt es neben dem Stack im Browser eine spezielle Warteschlange für die Arbeit mit dem sogenannten WebAPI. Die Funktionen aus dieser Warteschlange werden der Reihe nach erst dann ausgeführt, wenn der Stack vollständig geleert ist. Erst danach gelangen sie aus der Warteschlange auf den Stack zur Ausführung. Wenn sich im Stack derzeit auch nur ein Element befindet, können sie nicht auf den Stack gelangen. Genau aus diesem Grund ist der Aufruf von Funktionen per Timeout oft zeitlich ungenau, da die Funktion nicht aus der Warteschlange auf den Stack gelangen kann, solange dieser gefüllt ist.

Betrachten wir das folgende Beispiel und beschäftigen wir uns mit seiner schrittweisen "Ausführung". Außerdem schauen wir uns an, was dabei im System passiert.

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

1) Zunächst passiert nichts. Die Konsole des Browsers ist leer, der Aufruf-Stack ist leer.

2) Dann wird der Befehl console.log('Hi') zum Aufruf-Stack hinzugefügt.

3) Und er wird ausgeführt.

4) Danach wird console.log('Hi') aus dem Aufruf-Stack entfernt.

5) Jetzt kommen wir zum Befehl setTimeout(function cb1() {… }). Dieser wird zum Aufruf-Stack hinzugefügt.

6) Der Befehl setTimeout(function cb1() {… }) wird ausgeführt. Der Browser erstellt einen Timer, der Teil der Web-API ist. Er beginnt mit dem Countdown.

7) Der Befehl setTimeout(function cb1() {… }) hat seine Arbeit beendet und wird aus dem Aufruf-Stack entfernt.

8) Der Befehl console.log('Bye') wird zum Aufruf-Stack hinzugefügt.

9) Der Befehl console.log('Bye') wird ausgeführt.

10) Der Befehl console.log('Bye') wird aus dem Aufruf-Stack entfernt.

11) Nachdem mindestens 5000 ms vergangen sind, beendet der Timer seine Arbeit und legt das Callback cb1 in die Callback-Warteschlange.

12) Die Ereignisschleife nimmt die Funktion cb1 aus der Callback-Warteschlange und fügt sie dem Aufruf-Stack hinzu.

13) Die Funktion cb1 wird ausgeführt und fügt console.log('cb1') dem Aufruf-Stack hinzu.

14) Der Befehl console.log('cb1') wird ausgeführt.

15) Der Befehl console.log('cb1') wird aus dem Aufruf-Stack entfernt.

16) Die Funktion cb1 wird aus dem Aufruf-Stack entfernt.

Lassen Sie uns ein Beispiel in Aktion betrachten:

So, nun haben wir gesehen, wie Asynchronität in JavaScript umgesetzt wird. Lassen Sie uns nun kurz über die Evolution des asynchronen Codes sprechen.

Evolution des asynchronen Codes.

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

Asynchrones Programmieren, wie wir es in JavaScript kennen, kann nur mit Funktionen umgesetzt werden. Diese können wie jede andere Variable an andere Funktionen übergeben werden. So entstanden die Callbacks. Das ist spannend und macht Spaß, bis es sich in Traurigkeit und Frustration verwandelt. Warum? Weil:

Mit dem Anstieg der Komplexität des Codes verwandelt sich das Projekt schnell in schwer verständliche, mehrfach verschachtelte Blöcke - die sogenannte "Callback-Hölle".
Fehlerbehandlung kann leicht übersehen werden.
Ausdrücke können nicht mit return zurückgegeben werden.

Mit dem Aufkommen von Promises wurde die Situation etwas besser.

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('Fehler: ', e);
});

Mit der Einführung von Promise-Ketten wurde die Lesbarkeit des Codes verbessert.
Es gibt jetzt eine separate Methode zur Fehlerbehandlung.
Die Möglichkeit zur parallelen Ausführung mit Promise.all wurde hinzugefügt.
Verschachtelte Asynchronität können wir mit async/await lösen.

Promises haben jedoch ihre Einschränkungen. Zum Beispiel kann ein Promise ohne zusätzliche Maßnahmen nicht abgebrochen werden, und was am wichtigsten ist – er arbeitet mit einem einzigen Wert.

Jetzt sind wir sanft zum reaktiven Programmieren übergegangen. Müde? Zum Glück kann man sich einen Tee machen, nachdenken und dann weiterlesen. Ich mache unterdessen weiter.

Reaktives Programmieren — ist eine Programmierparadigma, die auf Datenströmen und der Verbreitung von Änderungen basiert. Lassen Sie uns genauer betrachten, was ein Datenstrom ist.

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

const eventsArray = [];

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

Stellen wir uns vor, wir haben ein Eingabefeld. Wir erstellen ein Array, und bei jedem Keyup-Event des Inputs speichern wir das Ereignis in unserem Array. Es ist wichtig zu beachten, dass unser Array nach Zeit sortiert ist – d.h. der Index späterer Ereignisse größer ist als der Index früherer Ereignisse. Ein solches Array stellt ein vereinfachtes Modell eines Datenstroms dar, aber es ist noch kein echter Strom. Um dieses Array sicher als Strom zu bezeichnen, muss es in der Lage sein, den Abonnenten irgendwie mitzuteilen, dass neue Daten eingegangen sind. So sind wir zur Definition eines Stroms gelangt.

Datenstrom

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

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

Strom— ist ein Array von zeitlich sortierten Daten, das darüber informieren kann, dass sich die Daten geändert haben. Stellen Sie sich vor, wie praktisch es wird, Code zu schreiben, bei dem eine Aktion mehrere Ereignisse an verschiedenen Stellen im Code auslösen muss. Wir abonnieren einfach den Strom, und er informiert uns selbst, wenn Änderungen auftreten. Und das kann die Bibliothek RxJs.

RxJS — ist eine Bibliothek zur Arbeit mit asynchronen und ereignisgesteuerten Programmen unter Verwendung beobachtbarer Sequenzen. Die Bibliothek bietet den Haupttyp Observable, mehrere Hilfstypen (Observer, Schedulers, Subjects) und Operatoren zur Verarbeitung von Ereignissen wie Sammlungen (map, filter, reduce, every und ähnliche aus dem JavaScript Array).

Lassen Sie uns die grundlegenden Konzepte dieser Bibliothek genauer betrachten.

Observable, Observer, Producer

Observable ist der erste grundlegende Typ, den wir betrachten werden. Diese Klasse enthält den Hauptteil der Implementierung von RxJs. Sie ist mit dem beobachtbaren Strom verbunden, auf den man sich mit der Methode subscribe abonnieren kann.

In Observable wird ein Hilfsmechanismus zur Erstellung von Updates implementiert, der so genannte Observer. Die Quelle der Werte für den Observer wird als Producerbezeichnet. Dies kann ein Array, ein Iterator, ein Websocket, ein Ereignis usw. sein. Man kann also sagen, dass Observable ein Vermittler zwischen Producer und Observer ist.

Observable verarbeitet drei Arten von Ereignissen des Observers:

next – neue Daten
error – einen Fehler, wenn die Sequenz aufgrund einer Ausnahme beendet wurde. Dieses Ereignis impliziert ebenfalls das Ende der Sequenz.
complete — Signal für den Abschluss der Sequenz. Dies bedeutet, dass keine neuen Daten mehr kommen werden.

Lass uns eine Demo anschauen:

Zu Beginn werden wir die Werte 1, 2, 3 verarbeiten, und nach 1 Sekunde erhalten wir 4 und beenden unseren Stream.

Gedanken laut

Und da wurde mir klar, dass es interessanter war, darüber zu erzählen, als es aufzuschreiben. 😀

Abonnement

Wenn wir ein Abonnement für den Stream erstellen, legen wir eine neue Klasse an subscription, die uns die Möglichkeit gibt, das Abonnement über die Methode unsubscribezu kündigen. Außerdem können wir Abonnements mit der Methode addgruppieren. Und es ist logisch, dass wir Streams mit removeentgruppieren können. Die Methoden add und remove akzeptieren eine andere Subscription als Eingabe. Ich möchte erwähnen, dass wenn wir uns abmelden, wir uns auch von allen untergeordneten Abonnements abmelden, als ob wir deren Methode unsubscribe aufgerufen hätten. Machen wir weiter.

Arten von Streams

HOT	COLD
Der Producer wird außerhalb des Observable erstellt.	Der Producer wird innerhalb des Observable erstellt.
Daten werden beim Erstellen des Observable übergeben.	Daten werden beim Abonnieren mitgeteilt.
Zusätzliche Logik für die Abmeldung ist erforderlich.	Der Stream endet selbstständig.
Verwendet eine Eins-zu-Viele-Beziehung.	Verwendet eine Eins-zu-Eins-Beziehung.
Alle Abonnements haben einen einheitlichen Wert.	Die Abonnements sind unabhängig
Daten können verloren gehen, wenn kein Abonnement besteht	Alle Werte des Streams werden für das neue Abonnement neu ausgegeben

Um eine Analogie zu ziehen, würde ich den heißen Stream mit einem Film im Kino vergleichen. Ab dem Zeitpunkt, an dem du angekommen bist, beginnst du mit dem Anschauen. Den kalten Stream würde ich mit einem Anruf beim technischen Support vergleichen. Jeder Anrufer hört sich die Ansage des Anrufbeantworters von Anfang bis Ende an, aber du kannst auflegen, indem du dich abmeldest.

Ich möchte darauf hinweisen, dass es auch die sogenannten warmen Streams gibt (eine solche Bezeichnung habe ich sehr selten gesehen und nur in ausländischen Communities) – das ist ein Stream, der sich von einem kalten Stream in einen heißen verwandelt. Die Frage ist – wo verwendet man das?)) Ich werde ein praktisches Beispiel anführen.

Ich arbeite mit Angular. Es nutzt aktiv RxJS. Für den Datenabruf vom Server erwarte ich einen kalten Stream und verwende diesen Stream im Template mit Hilfe des asyncPipe. Wenn ich diesen Pipe mehrfach verwende, wird jeder Pipe, zurückkommend auf die Definition des kalten Streams, die Daten vom Server anfordern, was, milde gesagt, seltsam ist. Wenn ich jedoch den kalten Stream in einen warmen Stream umwandle, erfolgt die Anfrage nur einmal.

Insgesamt ist das Verständnis der verschiedenen Stream-Arten für Anfänger ziemlich herausfordernd, aber wichtig.

Operatoren

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

Die Erweiterung der Möglichkeiten zur Arbeit mit Streams bieten uns Operatoren. Sie helfen, die Ereignisse, die in Observable ablaufen, zu steuern. Wir werden ein paar der beliebtesten betrachten. Detailliertere Informationen zu Operatoren finden Sie in den Links zur nützlichen Information.

Operatoren — of

Lassen Sie uns mit dem Hilfsoperator of beginnen. Er erstellt ein Observable basierend auf einem einfachen Wert.

Operatoren — filter

Der Filteroperator filter, wie der Name schon sagt, filtert das Signal des Streams. Wenn der Operator wahr zurückgibt, wird es weitergeleitet.

Operatoren — take

take — Gibt die Anzahl der Emissionen an, nach der der Stream beendet wird.

Operatoren — debounceTime

debounceTime — verwirft Emissionen, die im angegebenen Zeitintervall zwischen den Ausgaben liegen — gibt nach Ablauf des Zeitintervalls den letzten Wert aus.

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

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Gibt alle 1000 ms einen Wert aus
  setInterval(() => {
    observer.next(i++)
  }, 1000);

  // Gibt alle 1500 ms einen Wert aus
  setInterval(() => {
    observer.next(i++)
  }, 1500);
}).pipe(
  debounceTime(700),  // Wartet 700 ms auf Werte, bevor sie verarbeitet werden
  take(3)
);

Operatoren — takeWhile

Gibt Werte aus, solange takeWhile nicht false zurückgibt, danach wird vom Stream abgemeldet.

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

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Gibt alle 1000 ms einen Wert aus
  setInterval(() => {
    observer.next(i++)
  }, 1000);
}).pipe(
  takeWhile(producer => producer < 5)
);

Operatoren — combineLatest

Der kombinierte Operator combineLatest ähnelt etwas dem promise.all. Er vereint mehrere Streams in einen einzigen. Nachdem jeder Stream mindestens einen Emit erzeugt hat, erhalten wir die letzten Werte von jedem in Form eines Arrays. Danach, nach einem beliebigen Emit aus den kombinierten Streams, gibt er neue Werte zurück.

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Emitte einen Wert alle 1000ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Emitte einen Wert alle 750ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

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

Operatoren — zip

Zip — wartet auf einen Wert aus jedem Stream und bildet ein Array basierend auf diesen Werten. Wenn aus einem Stream kein Wert kommt, wird die Gruppe nicht gebildet.

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Emit value every 1000ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Emit value every 750ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

const observer_3 = Observable.create((observer) => {
  let i = 1;
  // Emit value every 500ms
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
});

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

Operators — forkJoin

forkJoin kombiniert ebenfalls Streams, gibt jedoch erst dann einen Wert aus, wenn alle Streams abgeschlossen sind (complete).

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Emit value every 1000ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
}).pipe(take(3));

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Emit value every 750ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
}).pipe(take(5));

const observer_3 = Observable.create((observer) => {
  let i = 1;
  // Emit value every 500ms
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
}).pipe(take(4));

forkJoin(observer_1, observer_2, observer_3);

Operators — map

Der Transformationsoperator map wandelt den Wert des Emit in einen neuen um.

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

Observable.create((observer) => {
  let i = 1;
  // Gibt alle 1000 ms einen Wert aus
  setInterval(() => {
    observer.next(i++);
  }, 1000);
}).pipe(
  map(x => x * 10),
  take(3)
);

Operatoren – share, tap

Der tap-Operator ermöglicht Nebenwirkungen, also Aktionen, die keinen Einfluss auf die Abfolge haben.

Der Utility-Operator share verwandelt einen kalten Stream in einen heißen.

Wir sind mit den Operatoren fertig. Kommen wir zum Subject.

Gedanken laut

Und hier habe ich eine Teepause gemacht. Diese Beispiele haben mich erschöpft 😀

Familie der Subjects

Die Familie der Subjects ist ein eindrucksvolles Beispiel für heiße Streams. Diese Klassen sind eine Art Hybrid, der sowohl als Observable als auch als Observer fungiert. Da Subject ein heißer Stream ist, muss man sich davon abmelden. Zu den Hauptmethoden gehören:

next – überträgt neue Daten in den Stream
error – Fehler und Beendigung des Streams
complete – Beendigung des Streams
subscribe – sich für den Stream anmelden
unsubscribe – sich vom Stream abmelden
asObservable – in einen Beobachter umwandeln
toPromise – in ein Promise umwandeln

Es gibt 4 bis 5 Typen von Subjects.

Gedanken laut

Im Stream sprach ich von 4, aber es scheint, dass sie noch einen hinzugefügt haben. Wie man sagt: Ein Leben lang lernen.

Einfaches Subject new Subject()– die einfachste Art von Subjects. Wird ohne Parameter erstellt. Überträgt Werte, die nur nach der Subscription empfangen werden.

BehaviorSubject new BehaviorSubject( defaultData ) – meiner Meinung nach die häufigste Art von Subjects. Nimmt einen Standardwert als Eingabe entgegen. Speichert immer die Daten des letzten Emits, die bei der Subscription übermittelt werden. Diese Klasse hat außerdem eine nützliche Methode value, die den aktuellen Wert des Streams zurückgibt.

ReplaySubject new ReplaySubject(bufferSize?: number, windowTime?: number) — Optional kann sie beim ersten Argument die Größe des Pufferwerts annehmen, den sie speichern soll, und beim zweiten das Zeitfenster, in dem wir die Änderungen benötigen.

AsyncSubject new AsyncSubject() — Bei der Subscription passiert nichts und der Wert wird nur bei complete zurückgegeben. Nur der letzte Wert des Streams wird zurückgegeben.

WebSocketSubject new WebSocketSubject(urlConfigOrSource: string | WebSocketSubjectConfig | Observable, destination?: Observer) — Die Dokumentation schweigt darüber und ich sehe es selbst zum ersten Mal. Wer weiß, was es macht, schreibt bitte, wir ergänzen.

So, das war alles, was ich heute besprechen wollte. Ich hoffe, die Informationen waren nützlich. Sie können die Literaturliste selbst im Tab "Nützliche Informationen" einsehen.