Asynchrone JavaScript-Programmierung (Callback, Promise, RxJs)

Hallo an alle. In Kontakt Omelnitsky Sergey. Vor nicht allzu langer Zeit leitete ich einen reaktiven Programmierstrom, in dem ich über Asynchronität in JavaScript sprach. Heute möchte ich dieses Material skizzieren.

Bevor wir jedoch mit dem Hauptmaterial beginnen, müssen wir eine Einführung machen. Beginnen wir also mit den Definitionen: Was ist der Stapel und die Warteschlange?

Ein Stapel ist eine Sammlung, deren Elemente nach dem Prinzip "last in, first out" LIFO empfangen werden

Eine Warteschlange ist eine Sammlung, deren Elemente nach dem Prinzip empfangen werden (FIFO „Wer zuerst kommt, mahlt zuerst“)

Ok, lass uns weitermachen.

JavaScript ist eine Single-Threaded-Programmiersprache. Dies bedeutet, dass es nur einen Ausführungsthread und einen Stapel gibt, in dem Funktionen zur Ausführung in die Warteschlange gestellt werden. Daher kann JavaScript zu einem bestimmten Zeitpunkt nur eine Operation ausführen, während andere Operationen darauf warten, dass sie den Stapel einschalten, bis sie aufgerufen werden.

Der Aufrufstapel ist eine Datenstruktur, die in einfachen Worten Informationen über die Stelle im Programm aufzeichnet, an der wir uns befinden. Wenn wir in eine Funktion gehen, setzen wir einen Datensatz darüber oben auf den Stapel. Wenn wir von der Funktion zurückkehren, ziehen wir das oberste Element aus dem Stapel und befinden uns dort, wo wir diese Funktion aufgerufen haben. Dies ist alles, was der Stapel tun kann. Und jetzt eine äußerst interessante Frage. Wie funktioniert dann Asynchronität in JavasScript?

Zusätzlich zum Stapel haben Browser eine spezielle Warteschlange für die Arbeit mit der sogenannten WebAPI. Funktionen aus dieser Warteschlange werden erst ausgeführt, nachdem der Stapel vollständig gelöscht wurde. Erst danach werden sie zur Ausführung aus der Warteschlange auf den Stapel geschoben. Befindet sich derzeit mindestens ein Element auf dem Stapel, können sie nicht auf den Stapel gelangen. Genau aus diesem Grund ist das Aufrufen von Funktionen nach Zeitüberschreitung häufig zeitlich nicht genau, da eine Funktion nicht von der Warteschlange zum Stapel wechseln kann, solange sie voll ist.

Betrachten Sie das folgende Beispiel und führen Sie die schrittweise „Ausführung“ durch. Sehen Sie auch, was im System passiert.

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

1) Bisher passiert nichts. Die Browserkonsole ist sauber, der Aufrufstapel ist leer.

2) Anschließend wird der Befehl console.log ('Hi') zum Aufrufstapel hinzugefügt.

3) Und es wird ausgeführt

4) Dann wird console.log ('Hi') vom Aufrufstapel entfernt.

5) Gehen Sie nun zum Befehl setTimeout (Funktion cb1 () {...}). Es wird dem Aufrufstapel hinzugefügt.

6) Der Befehl setTimeout (Funktion cb1 () {...}) wird ausgeführt. Der Browser erstellt einen Timer, der Teil der Web-API ist. Er wird den Countdown machen.

7) Der Befehl setTimeout (Funktion cb1 () {...}) wurde abgeschlossen und aus dem Aufrufstapel entfernt.

8) Der Befehl console.log ('Bye') wird dem Aufrufstapel hinzugefügt.

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

10) Der Befehl console.log ('Bye') wird aus dem Aufrufstapel entfernt.

11) Nach mindestens 5000 ms wird der Timer beendet und der cb1-Rückruf in die Rückrufwarteschlange gestellt.

12) Die Ereignisschleife nimmt c die Funktion cb1 aus der Rückrufwarteschlange und legt sie auf dem Aufrufstapel ab.

13) Die Funktion cb1 wird ausgeführt und fügt console.log ('cb1') zum Aufrufstapel hinzu.

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

15) Der Befehl console.log ('cb1') wird aus dem Aufrufstapel entfernt.

16) Die Funktion cb1 wird aus dem Aufrufstapel entfernt.

Schauen Sie sich ein Beispiel in der Dynamik an:

Hier haben wir untersucht, wie Asynchronität in JavaScript implementiert ist. Lassen Sie uns nun kurz über die Entwicklung des asynchronen Codes sprechen.

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

Asynchrone Programmierung, wie wir sie in JavaScript kennen, kann nur mit Funktionen implementiert werden. Sie können wie jede andere Variable an andere Funktionen übergeben werden. So wurden die Rückrufe geboren. Und es ist cool, lustig und provokativ, bis es sich in Traurigkeit, Sehnsucht und Traurigkeit verwandelt. Warum? Ja, alles ist einfach:

• Mit der zunehmenden Komplexität des Codes verwandelt sich das Projekt schnell in obskure, wiederholt verschachtelte Blöcke - „Callback Hell“.
• Fehlerbehandlung kann leicht übersehen werden.
• Sie können keine Ausdrücke mit return zurückgeben.

Mit Promise wurde es 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('error: ', e); }); 

• Es erschienen Versprechensketten, die die Lesbarkeit des Codes verbesserten
• Eine separate Fehlerüberwachungsmethode wurde angezeigt
• Jetzt können Sie mit Promise.all parallel ausführen
• Wir können verschachtelte Asynchronität mit async / await lösen

Aber Promis hat seine Grenzen. Zum Beispiel kann ein Versprechen, ohne mit einem Tamburin zu tanzen, nicht rückgängig gemacht werden, und vor allem funktioniert es mit einem Wert.

Nun, wir näherten uns reibungslos der reaktiven Programmierung. Bist du müde Das Gute ist, dass Sie ein paar Möwen machen, nachdenken und zurückkehren können, um weiterzulesen. Und ich werde weitermachen.

Reaktive Programmierung ist ein Programmierparadigma, das sich auf Datenflüsse und die Ausbreitung von Veränderungen konzentriert. Schauen wir uns einen Datenstrom genauer an.

 //     const input = ducument.querySelector('input'); const eventsArray = []; //      eventsArray input.addEventListener('keyup', event => eventsArray.push(event) ); 

Stellen Sie sich vor, wir haben ein Eingabefeld. Wir erstellen ein Array und speichern das Ereignis für jedes Keyup des Eingabeereignisses in unserem Array. In diesem Fall möchte ich darauf hinweisen, dass unser Array nach Zeit sortiert ist, d.h. Der Index späterer Ereignisse ist größer als der Index früherer Ereignisse. Ein solches Array ist ein vereinfachtes Datenstrommodell, aber noch kein Stream. Damit dieses Array sicher als Stream bezeichnet werden kann, muss es die Teilnehmer irgendwie darüber informieren können, dass es neue Daten erhalten hat. Wir kommen also zur Definition des Flusses.

Datenstrom

 const { interval } = Rx; const { take } = RxOperators; interval(1000).pipe( take(4) ) 

Ein Stream ist ein nach Zeit sortiertes Datenarray, das anzeigen kann, dass sich die Daten geändert haben. Stellen Sie sich nun vor, wie bequem es ist, Code zu schreiben, in dem Sie mehrere Ereignisse in verschiedenen Teilen des Codes in einer Aktion auslösen müssen. Wir abonnieren einfach den Stream und er wird uns wissen lassen, wann die Änderungen eintreten. Und die RxJs-Bibliothek kann dies tun.

RxJS ist eine Bibliothek für die Arbeit mit asynchronen und ereignisbasierten Programmen unter Verwendung beobachtbarer Sequenzen. Die Bibliothek bietet den Haupttyp von Observable , verschiedene Hilfstypen ( Observer, Scheduler, Subjects ) und Operatoren für die Arbeit mit Ereignissen wie bei Sammlungen ( Map, Filter, Reduce, alles und dergleichen aus JavaScript Array).

Schauen wir uns die Grundkonzepte dieser Bibliothek an.

Beobachtbar, Beobachter, Produzent

Observable ist der erste Grundtyp, den wir betrachten werden. Diese Klasse enthält den Großteil der RxJ-Implementierung. Es ist einem beobachtbaren Stream zugeordnet, den Sie beide mit der Subscribe-Methode abonnieren können.

Observable implementiert einen Hilfsmechanismus zum Erstellen von Updates, den sogenannten Observer . Die Quelle der Werte für Observer heißt Producer . Dies kann ein Array, ein Iterator, ein Web-Socket, eine Art Ereignis usw. sein. Wir können also sagen, dass Observable ein Dirigent zwischen Produzent und Beobachter ist.

Observable behandelt drei Arten von Observer-Ereignissen:

• next - neue Daten
• Fehler - Ein Fehler, wenn die Sequenz aufgrund einer Ausnahme beendet wurde. Dieses Ereignis beinhaltet auch die Vervollständigung der Sequenz.
• complete - Signal über den Abschluss der Sequenz. Dies bedeutet, dass keine neuen Daten vorhanden sind.

Schauen wir uns die Demo an:

Zu Beginn werden wir die Werte 1, 2, 3 und nach 1 Sek. Verarbeiten. Wir werden 4 bekommen und unseren Fluss beenden.

Abonnement

Wenn wir einen Stream abonnieren, erstellen wir eine neue Abonnementklasse , mit der wir uns mit der Methode zum Abbestellen abmelden können . Wir können Abonnements auch mit der Methode add gruppieren. Nun, es ist logisch, dass wir die Gruppierung der Threads mit remove aufheben können . Die Eingabemethoden zum Hinzufügen und Entfernen akzeptieren ein anderes Abonnement. Ich möchte darauf hinweisen, dass wir beim Abbestellen alle untergeordneten Abonnements abbestellen, als würden sie die Abmeldemethode aufrufen. Mach weiter.

Arten von Streams

Um eine Analogie zu geben, würde ich mir einen heißen Stream wie einen Film in einem Kino vorstellen. Zu welchem ​​Zeitpunkt kamen Sie von diesem Moment an und begannen zu sehen. Ich würde einen kalten Strom mit einem Anruf in diesen vergleichen. Unterstützung. Jeder Anrufer hört von Anfang bis Ende auf den Anrufbeantworter, aber Sie können auflegen, indem Sie sich abmelden.

Ich möchte darauf hinweisen, dass es immer noch sogenannte Warmströme gibt (eine solche Definition habe ich sehr selten und nur in fremden Gemeinden getroffen) - dies ist ein Strom, der sich von einem kalten in einen heißen Strom verwandelt. Es stellt sich die Frage - wo zu verwenden)) Ich werde ein Beispiel aus der Praxis geben.

Ich arbeite mit einem Winkel. Er benutzt aktiv rxjs. Um Daten auf den Server zu übertragen, erwarte ich einen kalten Stream und verwende diesen Stream in der Vorlage mit asyncPipe. Wenn ich diese Pipe mehrmals verwende, fordert jede Pipe, um zur Definition eines kalten Streams zurückzukehren, Daten vom Server an, was gelinde gesagt seltsam ist. Und wenn ich einen kalten Strom in einen warmen umwandle, wird die Anfrage einmal gestellt.

Im Allgemeinen ist das Verständnis der Form von Flüssen für Anfänger recht kompliziert, aber wichtig.

Betreiber

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

Operatoren bieten uns die Möglichkeit, mit Streams zu arbeiten. Sie helfen bei der Steuerung von Ereignissen, die im Observable auftreten. Wir werden einige der beliebtesten betrachten, und die Betreiber können über die Links in den nützlichen Informationen detaillierter gefunden werden.

Betreiber - von

Wir beginnen mit dem Hilfsoperator von. Es wird ein Observable basierend auf einem einfachen Wert erstellt.

Operatoren - Filter

Der Filteroperatorfilter filtert, wie der Name schon sagt, das Stream-Signal. Wenn der Operator true zurückgibt, wird weiter übersprungen.

Betreiber - nehmen

take - Nimmt den Wert der Anzahl der Emits an, nach denen der Stream endet.

Operatoren - debounceTime

debounceTime - verwirft die ausgegebenen Werte, die in den angegebenen Zeitraum zwischen den Ausgabedaten fallen. Nach Ablauf des Zeitintervalls wird der letzte Wert ausgegeben.

 const { Observable } = Rx; const { debounceTime, take } = RxOperators; Observable.create((observer) => { let i = 1; observer.next(i++); //     1000 setInterval(() => { observer.next(i++) }, 1000); //     1500 setInterval(() => { observer.next(i++) }, 1500); }).pipe( debounceTime(700), //  700     take(3) ); 

Operatoren - takeWhile

Es gibt Werte aus, bis takeWhile false zurückgibt. Danach wird der Stream abbestellt.

 const { Observable } = Rx; const { debounceTime, takeWhile } = RxOperators; Observable.create((observer) => { let i = 1; observer.next(i++); //     1000 setInterval(() => { observer.next(i++) }, 1000); }).pipe( takeWhile( producer => producer < 5 ) ); 

Operatoren - kombinierenLetzte

Der Operator kombinieren kombinierenLatest ähnelt etwas versprechen.all. Es kombiniert mehrere Threads zu einem. Nachdem jeder Thread mindestens eine Ausgabe durchgeführt hat, erhalten wir die letzten Werte von jedem in Form eines Arrays. Darüber hinaus ergeben sich nach jeder Emission aus den kombinierten Flüssen neue Werte.

 const { combineLatest, Observable } = Rx; const { take } = RxOperators; const observer_1 = Observable.create((observer) => { let i = 1; //     1000 setInterval(() => { observer.next('a: ' + i++); }, 1000); }); const observer_2 = Observable.create((observer) => { let i = 1; //     750 setInterval(() => { observer.next('b: ' + i++); }, 750); }); combineLatest(observer_1, observer_2).pipe(take(5)); 

Bediener - Reißverschluss

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

 const { zip, Observable } = Rx; const { take } = RxOperators; const observer_1 = Observable.create((observer) => { let i = 1; //     1000 setInterval(() => { observer.next('a: ' + i++); }, 1000); }); const observer_2 = Observable.create((observer) => { let i = 1; //     750 setInterval(() => { observer.next('b: ' + i++); }, 750); }); const observer_3 = Observable.create((observer) => { let i = 1; //     500 setInterval(() => { observer.next('c: ' + i++); }, 500); }); zip(observer_1, observer_2, observer_3).pipe(take(5)); 

Operatoren - forkJoin

forkJoin verkettet auch Threads, gibt jedoch nur Werte an, wenn alle Threads vollständig sind.

 const { forkJoin, Observable } = Rx; const { take } = RxOperators; const observer_1 = Observable.create((observer) => { let i = 1; //     1000 setInterval(() => { observer.next('a: ' + i++); }, 1000); }).pipe(take(3)); const observer_2 = Observable.create((observer) => { let i = 1; //     750 setInterval(() => { observer.next('b: ' + i++); }, 750); }).pipe(take(5)); const observer_3 = Observable.create((observer) => { let i = 1; //     500 setInterval(() => { observer.next('c: ' + i++); }, 500); }).pipe(take(4)); forkJoin(observer_1, observer_2, observer_3); 

Operatoren - Karte

Der Kartentransformationsoperator konvertiert den Ausgabewert in einen neuen.

 const { Observable } = Rx; const { take, map } = RxOperators; Observable.create((observer) => { let i = 1; //     1000 setInterval(() => { observer.next(i++); }, 1000); }).pipe( map(x => x * 10), take(3) ); 

Operatoren - teilen, tippen

Mit dem Tap-Operator können Sie Nebenwirkungen ausführen, dh alle Aktionen, die die Sequenz nicht beeinflussen.

Der Utility-Share-Betreiber kann es aus einem kalten Strom heiß machen.

Mit den Bedienern fertig. Fahren wir mit dem Thema fort.

Fachfamilie

Die Themenfamilie ist ein Paradebeispiel für heiße Streams. Diese Klassen sind eine Art Hybrid, die gleichzeitig als beobachtbar und beobachtend fungieren. Da es sich bei dem Betreff um einen heißen Stream handelt, müssen Sie ihn abbestellen. Wenn wir über die grundlegenden Methoden sprechen, dann ist dies:

• next - Übertragung neuer Daten in den Stream
• Fehler - Fehler und Beendigung des Streams
• vollständig - Beendigung des Streams
• Abonnieren - Abonnieren Sie den Stream
• Abbestellen - Abbestellen vom Stream
• asObservable - verwandeln Sie sich in einen Beobachter
• toPromise - verwandelt sich in ein Versprechen

Zuweisen 4 5 Arten von Themen.

Einfaches Thema new Subject() ist die einfachste Art von Thema. Es wird ohne Parameter erstellt. Übergibt Werte, die erst nach dem Abonnement kamen.

BehaviorSubject new BehaviorSubject( defaultData<T> ) - meiner Meinung nach die häufigste Art von Thema. Die Eingabe akzeptiert einen Standardwert. Es werden immer die Daten der letzten Ausgabe gespeichert, die beim Abonnieren übertragen werden. Diese Klasse verfügt auch über eine nützliche Wertemethode, die den aktuellen Wert des Streams zurückgibt.

ReplaySubject new ReplaySubject(bufferSize?: number, windowTime?: number) - Die Eingabe kann optional das erste Argument als Größe des new ReplaySubject(bufferSize?: number, windowTime?: number) akzeptieren, den es in sich selbst speichert, und das zweite Mal, während dem Änderungen erforderlich sind.

AsyncSubject new AsyncSubject() - Beim Abonnieren passiert nichts, und der Wert wird erst zurückgegeben, wenn er abgeschlossen ist. Es wird nur der letzte Stream-Wert zurückgegeben.

WebSocketSubject new WebSocketSubject(urlConfigOrSource: string | WebSocketSubjectConfig<T> | Observable<T>, destination?: Observer<T>) - Die Dokumentation enthält new WebSocketSubject(urlConfigOrSource: string | WebSocketSubjectConfig<T> | Observable<T>, destination?: Observer<T>) dazu und ich sehe sie zum ersten Mal. Wer weiß, was er tut, schreibt, ergänzt.

Fuf. Nun, hier haben wir alles überlegt, was ich heute erzählen wollte. Ich hoffe diese Information war hilfreich. Sie können sich mit der Referenzliste auf der Registerkarte Nützliche Informationen vertraut machen.

Eine nützliche Information

• Stream Link
• Funktionsweise von JS: Übersicht über die Engine, Laufzeitmechanismen, Aufrufstapel
• Funktionsweise von JS: Ereignisschleife, Async und fünf Möglichkeiten zur Verbesserung des Codes mithilfe von async / await
• Funktionsweise der Ereignisschleife in JavaSript
• Asynchrone JavaScript-Evolution
• Was ist RxJS und warum ist es nützlich, darüber Bescheid zu wissen?
  
• Praktische Anwendung von RxJS
• RxJS Observables Tutorial - Erstellen und Abonnieren von Observables
• RXJS: Heiße und kalte Observable
• Klassen, Funktionen zum Erstellen eines Observable. Betreiber
• RxJS-Operatoren am Beispiel
• API-Liste
• Sorten von Themen und Zeitplänen in RxJS