Programmation asynchrone en JavaScript. (Callback, Promise, RxJs)

Bonjour à tous. Ici, Sergey Omelnitsky. Il n'y a pas si longtemps, j'ai animé un stream sur la programmation réactive, où j'ai parlé de l'asynchrone en JavaScript. Aujourd'hui, je voudrais résumer ce matériel.

Mais avant de commencer le contenu principal, nous devons faire une introduction. Alors, commençons par les définitions : qu'est-ce qu'une pile et une file d'attente ?
Pile — c'est une collection dont les éléments sont obtenus selon le principe « dernier entré, premier sorti » LIFO.
File d'attente — c'est une collection dont les éléments sont reçus selon le principe « premier entré, premier sorti » FIFO.

D'accord, continuons.

JavaScript est un langage de programmation à thread unique. Cela signifie qu'il n'y a qu'un seul fil d'exécution et une seule pile, dans laquelle les fonctions sont placées en attente d'exécution. Par conséquent, à un moment donné, JavaScript ne peut exécuter qu'une seule opération, les autres opérations attendant leur tour dans la pile jusqu'à ce qu'elles soient appelées.

Pile d'appels — c'est une structure de données qui, en termes simples, enregistre des informations sur l'endroit où nous en sommes dans le programme. Lorsque nous entrons dans une fonction, nous ajoutons un enregistrement à son sujet en haut de la pile. Lorsque nous revenons de la fonction, nous retirons l'élément du haut de la pile et nous nous retrouvons là d'où nous avons appelé cette fonction. C'est tout ce que fait la pile. Et maintenant, une question très intéressante. Comment l'asynchrone fonctionne-t-il alors en JavaScript ?

En réalité, en plus de la pile, il y a dans les navigateurs une file d'attente spéciale pour travailler avec ce que l'on appelle le WebAPI. Les fonctions de cette file d'attente ne s'exécutent qu'après que la pile est complètement nettoyée. Ce n'est qu'après cela qu'elles passent de la file d'attente à la pile pour exécution. Si la pile contient au moins un élément, elles ne peuvent pas entrer dans la pile. C'est exactement pour cette raison que l'appel de fonctions par timeout peut souvent ne pas être précis en termes de temps, car la fonction ne peut pas entrer de la file d'attente dans la pile tant qu'elle est pleine.

Considérons l'exemple suivant et analysons son « exécution » étape par étape. De plus, regardons ce qui se passe dans le système.

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

1) Rien ne se passe encore. La console du navigateur est vide, la pile d'appels est vide.

2) Ensuite, la commande console.log('Hi') est ajoutée à la pile d'appels.

3) Et elle s'exécute.

4) Ensuite, console.log('Hi') est supprimé de la pile d'appels.

5) Nous passons maintenant à la commande setTimeout(function cb1() {… }). Elle est ajoutée à la pile d'appels.

6) La commande setTimeout(function cb1() {… }) est exécutée. Le navigateur crée un minuteur qui fait partie de l'API Web. Il comptera à rebours.

7) La commande setTimeout(function cb1() {… }) a terminé son travail et est supprimée de la pile d'appels.

8) La commande console.log('Bye') est ajoutée à la pile d'appels.

9) La commande console.log('Bye') est exécutée.

10) La commande console.log('Bye') est supprimée de la pile d'appels.

11) Après au moins 5000 ms, le minuteur termine son travail et place le callback cb1 dans la file d'attente des callbacks.

12) La boucle d'événements prend la fonction cb1 de la file d'attente des callbacks et l'ajoute à la pile d'appels.

13) La fonction cb1 est exécutée et ajoute console.log('cb1') à la pile d'appels.

14) La commande console.log('cb1') est exécutée.

15) La commande console.log('cb1') est supprimée de la pile d'appels.

16) La fonction cb1 est supprimée de la pile d'appels.

Regardons un exemple en dynamique :

Eh bien, nous avons vu comment l'asynchronicité est implémentée en JavaScript. Maintenant, discutons brièvement de l'évolution du code asynchrone.

L'évolution du code asynchrone.

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

La programmation asynchrone telle que nous la connaissons en JavaScript ne peut être réalisée qu'avec des fonctions. Celles-ci peuvent être transmises comme n'importe quelle autre variable à d'autres fonctions. C'est ainsi qu'ont été créés les callbacks. Et c'est bien, amusant et joyeux, jusqu'à ce que cela se transforme en tristesse, mélancolie et chagrin. Pourquoi ? C'est simple:

• Avec l'augmentation de la complexité du code, le projet devient rapidement des blocs imbriqués difficiles à comprendre — «callback hell».
• La gestion des erreurs peut facilement être négligée.
• On ne peut pas retourner des expressions avec return.

Avec l'apparition de Promise, la situation est devenue un peu meilleure.

new Promise(function(resolve, reject) {
    setTimeout(() => resolve(1), 2000);

}).then((result) => {
    alert(result);
    return result + 2;

}).then((result) => {
    throw new Error('ÉCHEC ICI');
    alert(result);
    return result + 2;

}).then((result) => {
    alert(result);
    return result + 2;

}).catch((e) => {
    console.log('erreur: ', e);
});

• Des chaînes de promesses sont apparues, ce qui a amélioré la lisibilité du code.
• Un méthode distincte de gestion des erreurs est apparue.
• Il est désormais possible d'exécuter des tâches en parallèle avec Promise.all
• Nous pouvons résoudre l'asynchronicité imbriquée avec async/await

Cependant, il y a des limitations avec les promesses. Par exemple, une promesse ne peut pas être annulée sans une manipulation compliquée, et surtout, elle ne fonctionne qu'avec une seule valeur.

Eh bien, nous avons tranquillement abordé la programmation réactive. Fatigué ? Heureusement, vous pouvez aller préparer une tasse de thé, réfléchir et revenir lire la suite. Pendant ce temps, je vais continuer.

Programmation réactive — une paradigme de programmation axée sur les flux de données et la diffusion des modifications. Analysons plus en détail ce qu'est un flux de données.

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

const eventsArray = [];

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

Imaginons que nous avons un champ de saisie. Nous créons un tableau, et à chaque événement keyup de l'input, nous allons sauvegarder cet événement dans notre tableau. Il est à noter que notre tableau est trié par temps, c'est-à-dire que l'index des événements plus récents est supérieur à l'index des événements plus anciens. Ce tableau représente un modèle simplifié d'un flux de données, mais ce n'est pas encore un flux. Pour que ce tableau puisse être désigné comme un flux, il doit être capable d'informer d'une manière ou d'une autre les abonnés qu'il a reçu de nouvelles données. Ainsi, nous sommes parvenus à la définition d'un flux.

Flux de données

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

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

Flux  — est un tableau de données, trié par temps, qui peut signaler que les données ont changé. Imaginez à quel point il devient pratique d'écrire du code qui nécessite d'appeler plusieurs événements à différents endroits du code. Nous nous abonnissons simplement au flux, et il nous informera lui-même des modifications. C'est ce que fait la bibliothèque RxJs.

RxJS  — une bibliothèque pour travailler avec des programmes asynchrones et basés sur des événements utilisant des séquences observables. La bibliothèque fournit le type principal Observable, plusieurs types auxiliaires (Observer, Schedulers, Subjects) et des opérateurs pour travailler avec des événements comme avec des collections (map, filter, reduce, every et similaires du tableau JavaScript).

Passons en revue les concepts de base de cette bibliothèque.

Observable, Observer, Producer

Observable — le premier type de base que nous allons examiner. Cette classe contient la majeure partie de l'implémentation de RxJs. Elle est liée à un flux observable, auquel on peut s'abonner via la méthode subscribe.

Un mécanisme auxiliaire pour créer des mises à jour est implémenté dans Observable, appelé Observer. La source des valeurs pour l'Observer est appelée Producteur. Cela peut être un tableau, un itérateur, un web socket, un événement, etc. On peut donc dire que l'observable agit comme un intermédiaire entre le Producteur et l'Observer.

Observable gère trois types d'événements de l'Observer :

• next – nouvelles données
• error – une erreur, si la séquence se termine en raison d'une exception. Cet événement implique aussi la fin de la séquence.
• complete — un signal indiquant la fin de la séquence. Cela signifie qu'il n'y aura plus de nouvelles données.

Voyons une démo :

Au début, nous traiterons les valeurs 1, 2, 3, puis après 1 seconde, nous obtiendrons 4 et terminerons notre flux.

Réflexion

Et là, j'ai réalisé qu'il était plus intéressant de raconter que d'écrire à ce sujet. 😀

Subscription

Lorsque nous nous abonnons à un flux, nous créons une nouvelle classe subscription, qui nous permet d'annuler l'abonnement grâce à la méthode unsubscribe. Nous pouvons également grouper les abonnements à l'aide de la méthode add. Il est logique que nous puissions dégroupper les flux grâce à remove. Les méthodes add et remove acceptent comme argument un autre abonnement. Il est important de noter que lorsque nous nous désinscrivons, cela désinscrit également tous les abonnements enfants comme si nous avions également appelé la méthode unsubscribe sur eux. Continuons.

Types de flux

Pour faire une analogie, je représenterais un flux chaud comme un film au cinéma. Peu importe à quel moment tu arrives, c'est à partir de ce moment que tu commences à regarder. Un flux froid pourrait être comparé à un appel au support technique. Quiconque appelle écoute l'enregistrement du répondeur de bout en bout, mais tu peux raccrocher avec la méthode unsubscribe.

Je voudrais souligner qu'il existe également ce qu'on appelle des flux chauds (une définition que j'ai rencontrée très rarement et uniquement dans des communautés étrangères) — c'est un flux qui se transforme d'un flux froid en un flux chaud. La question se pose : où les utiliser ? Je vais donner un exemple pratique.

Je travaille avec Angular. Il utilise activement RxJS. Pour obtenir des données du serveur, j'attends un flux froid et j'utilise ce flux dans le modèle avec asyncPipe. Si j'utilise ce pipe plusieurs fois, alors, pour revenir à la définition du flux froid, chaque pipe va demander des données au serveur, ce qui est, pour le dire poliment, étrange. Mais si je transforme un flux froid en flux chaud, la demande se fera une seule fois.

En fait, la compréhension des types de flux est assez complexe pour les débutants, mais c'est important.

Opérateurs

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

Nous pouvons étendre les possibilités de travail avec les flux grâce aux opérateurs. Ils aident à contrôler les événements qui se produisent dans un Observable. Nous allons examiner quelques-uns des plus populaires, et pour plus de détails sur les opérateurs, vous pouvez consulter les liens dans les informations utiles.

Opérateurs — of

Commençons par l'opérateur auxiliaire of. Il crée un Observable à partir d'une valeur simple.

Opérateurs — filter

L'opérateur de filtrage filter, comme son nom l'indique, filtre le signal du flux. Si l'opérateur retourne vrai, il passe à la suite.

Opérateurs — take

take — Prend le nombre d'émissions après lequel il termine le flux.

Opérateurs — debounceTime

debounceTime — rejette les valeurs émises qui se produisent dans un intervalle de temps spécifié entre les sorties — après l'intervalle de temps écoulé, il émet la dernière valeur.

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

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Émet une valeur toutes les 1000ms
  setInterval(() => {
    observer.next(i++)
  }, 1000);

  // Émet une valeur toutes les 1500ms
  setInterval(() => {
    observer.next(i++)
  }, 1500);
}).pipe(
  debounceTime(700),  // Attend 700ms avant de traiter la valeur
  take(3)
);  

Opérateurs — takeWhile

Émet des valeurs tant que takeWhile ne retourne pas false, après quoi il se désabonne du flux.

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

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Émet une valeur toutes les 1000 ms
  setInterval(() => {
    observer.next(i++)
  }, 1000);
}).pipe(
  takeWhile(producer => producer < 5)
);  

Opérateurs — combineLatest

L'opérateur combiné combineLatest est similaire à promise.all. Il regroupe plusieurs flux en un seul. Une fois que chaque flux a émis au moins une valeur, nous recevons les dernières valeurs de chacun sous forme de tableau. Ensuite, après n'importe quelle émission des flux combinés, il renverra de nouvelles valeurs.

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Émet une valeur toutes les 1000 ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Émet une valeur toutes les 750 ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

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

Opérateurs — zip

Zip attend une valeur de chaque flux et forme un tableau basé sur ces valeurs. Si aucune valeur n'est reçue d'un flux, le groupe ne sera pas formé.

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Émet une valeur toutes les 1000 ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Émet une valeur toutes les 750 ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

const observer_3 = Observable.create((observer) => {
  let i = 1;
  // Émet une valeur toutes les 500 ms
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
});

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

Opérateurs — forkJoin

forkJoin regroupe également les flux, mais il émet une valeur uniquement lorsque tous les flux sont terminés (complétés).

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Émet une valeur toutes les 1000 ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
}).pipe(take(3));

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Émet une valeur toutes les 750 ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
}).pipe(take(5));

const observer_3 = Observable.create((observer) => {
  let i = 1;
  // Émet une valeur toutes les 500 ms
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
}).pipe(take(4));

forkJoin(observer_1, observer_2, observer_3);

Opérateurs — map

L'opérateur de transformation map convertit la valeur émise en une nouvelle.

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

Observable.create((observer) => {
  let i = 1;
  // Émet une valeur toutes les 1000 ms
  setInterval(() => {
    observer.next(i++);
  }, 1000);
}).pipe(
  map(x => x * 10),
  take(3)
);

Opérateurs – share, tap

L'opérateur tap permet d'exécuter des effets secondaires, c'est-à-dire toute action qui n'affecte pas la séquence.

L'opérateur utilitaire share peut transformer un flux froid en un flux chaud.

Nous avons terminé avec les opérateurs. Passons au Subject.

Réflexion

Et là, je suis allé prendre un thé. Ces exemples m'ont fatigué 😀

La famille des subjects

La famille des subjects est un exemple frappant de flux chauds. Ces classes sont en quelque sorte des hybrides, agissant à la fois comme observable et observer. Étant donné que le subject est un flux chaud, il est nécessaire de s'y désabonner. En ce qui concerne les principales méthodes, les voici :

• next – envoi de nouvelles données dans le flux
• error – erreur et terminaison du flux
• complete – terminaison du flux
• subscribe – s'abonner au flux
• unsubscribe – se désabonner du flux
• asObservable – transformé en observateur
• toPromise – transformé en promesse

On distingue 4 ou 5 types de subjects.

Réflexion

J'ai parlé de 4 lors du stream, mais apparemment ils en ont ajouté un de plus. Comme on dit, on vit et on apprend.

Subject simple new Subject()– le type de subject le plus simple. Créé sans paramètres. Transmet les valeurs reçues uniquement après l'abonnement.

BehaviorSubject new BehaviorSubject(defaultData) – à mon avis, le type de subject le plus couramment utilisé. Prend en entrée une valeur par défaut. Toujours conserve les données de la dernière émission, qui sont transmises lors de l'abonnement. Cette classe possède également la méthode utile value, qui retourne la valeur actuelle du flux.

ReplaySubject new ReplaySubject(bufferSize?: number, windowTime?: number) — En entrée, il peut optionnellement prendre comme premier argument la taille du buffer des valeurs qu'il va stocker, et comme deuxième le temps pendant lequel nous avons besoin des modifications.

AsyncSubject new AsyncSubject() — lors de l'abonnement, rien ne se passe, et la valeur sera retournée uniquement lors de complete. Seule la dernière valeur du flux sera retournée.

WebSocketSubject new WebSocketSubject(urlConfigOrSource: string | WebSocketSubjectConfig | Observable, destination?: Observer) — La documentation est silencieuse à son sujet et je le vois pour la première fois. Qui sait ce qu'il fait, écrivez, nous compléterons.

Ouf. Eh bien, nous avons couvert tout ce que je voulais dire aujourd'hui. J'espère que cette information a été utile. Vous pouvez consulter la bibliographie dans l'onglet informations utiles.

Informations utiles

• https://vk.com/frontend_practice?w=wall-182297870_931 — lien vers le stream
• https://habr.com/ru/company/ruvds/blog/337042/ — Comment fonctionne JS : aperçu du moteur, des mécanismes d'exécution, de la pile d'appels
• https://habr.com/ru/company/ruvds/blog/340508 — Comment fonctionne JS : boucle d'événements, asynchronisme et cinq façons d'améliorer le code avec async / await
• http://vanilla-js.blogspot.com/2014/09/event-loop-javasript.html — Comment fonctionne la boucle d'événements en JavaScript
• https://tproger.ru/translations/asynchronous-javascript/ — L'évolution de JavaScript asynchrone
• https://habr.com/ru/company/ruvds/blog/341880/ — Qu'est-ce que RxJS et pourquoi est-il utile de le connaître
• https://medium.com/ngx/practical-use-rxjs-81aaab57045c — Applications pratiques de RxJS
• https://coursetro.com/posts/code/148/RxJS-Observables-Tutorial—Creating-&-Subscribing-to-Observables — Tutoriel RxJS Observables — Création et abonnement aux Observables
• https://slides.com/zenabsaleem/rxjs-hot-cold-observables#/ — RXJS : Observables chauds et froids
• https://clck.ru/FiPgB — Classes, fonctions pour créer des Observable. Opérateurs.
• https://www.learnrxjs.io/operators/ — Opérateurs RxJS par exemple
• https://rxjs.dev/api — Liste API
• https://clck.ru/FkS8Y — Types de Subject et Scheduling dans RxJS

Source : habr.com