Cet article a été examiné par Florian Rappl et Moritz Kr?ger. Merci à tous les pairs examinateurs de SitePoint pour avoir rendu le contenu de SitePoint parfait!

à mesure que l'intérêt pour la programmation réactive fonctionnelle (FRP) augmente, RXJS est devenu l'une des bibliothèques JavaScript les plus populaires de ce paradigme. Dans cet article, nous explorerons les dix premières fonctions incontournables dans RXJS.

Remarque: Cet article suppose que vous connaissez les bases des RXJ, comme décrit dans l'article "commen?ant par la programmation réactive fonctionnelle avec RXJS".

points clés

• RXJS exploite des objets observables similaires aux tableaux remplis dans le temps pour faciliter la programmation réactive fonctionnelle (FRP), permettant une gestion des erreurs plus déclarative et puissante dans les applications.
• Les opérations principales des flux simples dans RXJS, tels que map(), filter(), reduce(), et take(), les opérations du tableau miroir, mais sont appliquées aux flux de nombres qui émettent des valeurs au fil du temps.
• Des fonctions spéciales telles que flatMap() et switch() sont essentielles pour gérer les structures de données complexes et gérer plusieurs flux séparément, ce qui les rend essentiels pour les taches de programmation réactives avancées.
Les opérateurs
• tels que concat(), merge() et combineLatest() peuvent être utilisés pour combiner efficacement plusieurs flux, chacun jouant un r?le différent dans la gestion des flux et la synchronisation des données.
La fonction
• takeUntil() fournit un mécanisme basé sur des conditions externes à se désabonner, qui illustre la flexibilité des RXJ dans le contr?le des flux et la gestion des ressources.

Programmation réactive

La programmation réactive est un paradigme de programmation qui prend le flux de données appelé objets observables comme son unité de programmation de base.

Stream - ou objets observables dans le jargon RXJS - Alike des écouteurs d'événements: les deux attendent que quelque chose se produise et vous informe quand cela se produit. Une série de notifications asynchrones obtenues auprès de l'auditeur OnClick est un parfait exemple de flux de données.

En d'autres termes, l'objet observable n'est rien de plus qu'un tableau rempli au fil du temps.

Les éléments de ce tableau peuvent provenir de presque n'importe où: système de fichiers, événements DOM, appels API et même données synchrones converties telles que les tableaux. Fondamentalement, la programmation réactive n'est rien de plus que d'utiliser des objets observables comme blocs de construction de programmes.

relation avec le tableau

Les tableaux

sont simples car leur contenu est définitif, sauf si c'est explicitement modifié. En ce sens, il n'y a pas de temporalité essentielle dans un tableau.

En revanche, les objets observables sont définis par le temps. Au plus, vous pouvez savoir que le flux a re?u jusqu'à présent [1, 2, 3]. Vous ne pouvez pas être s?r que vous obtenez 4 - ou pas - et c'est la source de données, pas votre programme, qui le détermine.

La relation entre les flux et les tableaux est si profonde que la plupart des extensions réactives proviennent du monde de la programmation fonctionnelle, où les opérations de liste sont centrales.

familier avec rxjs

Considérons les applications de taches communes. Voyons la question de savoir comment afficher le nom de la tache inachevée d'un utilisateur à l'aide de RXJS:

const task_stream =
  // 創(chuàng)建所有數(shù)據(jù)庫中任務的流
  getTasks().
    // 只獲取此用戶的任務
    filter((task) => task.user_id == user_id).
    // 獲取未完成的任務
    filter((task) => !task.completed).
    // 只獲取任務名稱
    map((task) => task.name)

/* 任務如下所示：
   task = {
    user_id   : number,
    completed : boolean,
    name      : string
   }
 */

Jusqu'à présent, ce n'est qu'une extension du tableau, mais il démontre le style fonctionnel de la programmation réactive.

La nature déclarative devient claire en ajoutant des fonctions plus complexes, "monde réel". Supposons que nous voulons:

• lancez la demande en réponse au choix de l'utilisateur de voir les taches terminées ou inachevées;
Envoyez une seule fois par seconde à la dernière sélection pour éviter de gaspiller la bande passante lorsque l'utilisateur modifie rapidement la sélection;
• réessayer jusqu'à trois demandes ratées; et
• Repaindre la vue uniquement si le serveur envoie une réponse différente de la dernière fois.

Décompose pas les pas:

const task_stream =
  parameter_stream.
    debounce(1000).
    map((parameter) => {
      getTasks().
        retry(3).
        filter((task) => task.user_id === user_id).
        filter((task) => task.completed === parameter).
        map((task)    => task.name)
    }).
    flatMap(Rx.Observable.from).
    distinctUntilChanged().
    update()

Paramètre_stream nous dit si l'utilisateur veut des taches terminées ou inachevées et stocke la sélection dans le paramètre;

Debounce () Assurez-vous que nous nous concentrons uniquement sur le dernier clic par seconde;
• La partie autour de getTasks () est la même qu'auparavant;
DistinTUnTilchanged () garantit que nous suivons la réponse du serveur différemment de la dernière fois;
• Update () est responsable de la mise à jour de l'interface utilisateur pour refléter ce que nous obtenons du serveur.
Gestion de débouchement, de réessayer et de "distincte jusqu'à ce que la logique" dans les styles impératifs basés sur un rappel est efficace, mais il est à la fois fragile et complexe.
La clé est que la programmation utilisant RXJS permet:

Programme déclaratif;

Système évolutif; et

Gestion des erreurs simples, directes et puissantes.

1. En train de parcourir les dix premières fonctions incontournables de RXJ, nous rencontrerons chacune des fonctions dans les exemples ci-dessus.
2. Fonctionnement d'écoulement simple
3. Les fonctions de base d'un flux simple (un flux qui émet des valeurs simples, comme une cha?ne) comprend:

map ()

filtre ()

réduire ()

Take () / Take What ()
• En dehors de Take () et Take What (), ceux-ci sont similaires aux fonctions de tableau d'ordre supérieur de JavaScript.
• Nous appliquerons ces fonctions en résolvant un exemple de problème: trouver tous les utilisateurs dans la base de données avec des sites Web .com ou .org et calculer la durée moyenne de leurs noms de site Web.
• JSONPlaceHolder servira de source d'utilisateurs. Il s'agit d'une représentation JSON des données utilisateur que nous utiliserons.

1.

L'utilisation de map () sur les objets observables est le même que l'utiliser sur un tableau. It:

Acceptez le rappel comme paramètre;

l'exécuter sur chaque élément du tableau que vous appelez; et

Renvoie un nouveau tableau où chaque élément du tableau d'origine est remplacé par le résultat produit par le rappel dessus.

La seule différence lors de l'utilisation de MAP () sur les objets observables est:

Il renvoie un nouvel objet observable, pas un nouveau tableau; et
Il s'exécute lorsque l'objet observable émet un nouveau projet, plut?t que tout cela immédiatement.

Nous pouvons utiliser MAP () pour convertir nos flux de données utilisateur en une liste qui ne contient que les noms de leur site Web:

const task_stream =
  // 創(chuàng)建所有數(shù)據(jù)庫中任務的流
  getTasks().
    // 只獲取此用戶的任務
    filter((task) => task.user_id == user_id).
    // 獲取未完成的任務
    filter((task) => !task.completed).
    // 只獲取任務名稱
    map((task) => task.name)

/* 任務如下所示：
   task = {
    user_id   : number,
    completed : boolean,
    name      : string
   }
 */

Ici, nous utilisons MAP pour remplacer chaque objet utilisateur dans le flux entrant avec le site Web de chaque utilisateur.

RXJS vous permet également d'appeler Map () comme select (). Les deux noms se réfèrent à la même fonction.

2.

comme map (), filter () joue à peu près le même r?le sur des objets observables que sur les tableaux. Pour trouver chaque utilisateur avec une adresse de site Web .NET ou .org, nous pouvons écrire ceci:

const task_stream =
  parameter_stream.
    debounce(1000).
    map((parameter) => {
      getTasks().
        retry(3).
        filter((task) => task.user_id === user_id).
        filter((task) => task.completed === parameter).
        map((task)    => task.name)
    }).
    flatMap(Rx.Observable.from).
    distinctUntilChanged().
    update()

Cela ne sélectionnera que les utilisateurs dont le site Web se termine par "net" ou "org".

Filter () a également un alias où ().

3.

réduction () nous permet d'utiliser toutes les valeurs uniques et de les convertir en un seul résultat.

réduction () est souvent l'opération de liste de base la plus déroutante, car contrairement à Filter () ou Map (), son comportement varie par utilisation.

Généralement, Reduce () prend une collecte de valeurs et les convertit en un seul point de données. Dans notre exemple, nous lui fournirons un flux de noms de site Web et utiliserons réduction () pour convertir ce flux en un objet qui calcule la somme du nombre de sites Web que nous avons trouvés et de sa longueur de nom.

Ici, nous simplifions le flux vers un seul objet, qui suit:

source.
  map((user) => user.website)

combien de sites avons-nous vu; et

1. Longueur totale de tous les noms.
N'oubliez pas que réduction () renvoie le résultat uniquement lorsque l'objet observable source est terminé. Si vous souhaitez conna?tre l'état de l'accumulateur chaque fois que le flux re?oit un nouvel élément, utilisez à la place Scan ().

4.

Take () et Take While () complètent les fonctions de base des flux simples.

prendre (n) lire n valeurs du flux et désabonner.

Nous pouvons utiliser Scan () pour émettre notre objet chaque fois que nous recevons le site Web, et prendre uniquement () les deux premières valeurs.

RXJS fournit également Take While (), qui vous permet d'obtenir des valeurs avant l'établissement d'un certain test booléen. Nous pouvons utiliser Take What () pour écrire le flux ci-dessus comme ceci:

source.
  map((user) => user.website).
  filter((website) => (website.endsWith('net') || website.endsWith('org'));
})

Fonctionnement d'écoulement d'ordre élevé

source.
  map((user) => user.website).
  filter((website) => (website.endsWith('net') || website.endsWith('org'))).
  reduce((data, website) => {
    return {
      count       : data.count += 1,
      name_length : data.name_length += website.length
    }
  }, { count : 0, name_length : 0 })

Ces fonctions sont presque les mêmes que les opérations de liste familières, sauf qu'elles fonctionnent sur des objets observables plut?t que sur des tableaux.

"[i] f Vous savez comment programmer contre les tableaux à l'aide du tableau # extras, alors vous savez déjà comment utiliser RXJS!"

Tout comme un tableau peut contenir des données plus complexes que les valeurs simples (comme un tableau ou un objet), les objets observables peuvent également émettre des données d'ordre supérieur, telles que la promesse ou d'autres objets observables. C'est là que davantage d'outils professionnels entrent en jeu.

5.

… En fait, nous l'utilisons déjà!

Lorsque nous définissons le flux source, nous appelons FromPromis () et FlatMap ():

Ceci utilise trois nouveaux mécanismes:

provenant de promesse;

source.
  map((user) => user.website).
  filter((website) => (website.endsWith('net') || website.endsWith('org'))).
  scan((data, website) => {
      return {
        count       : data.count += 1,
        name_length : data.name_length += website.length
      }
    }, { count : 0, name_length : 0 }).
  take(2);

rx.observable.from; et

Flatmap.

Objet observable de promesse
Promise représente une seule valeur future que nous obtiendrons de manière asynchrone - par exemple, le résultat d'un appel au serveur.
3. Une caractéristique déterminante de

   Promise est qu'elle ne représente qu'une valeur future. Il ne peut pas renvoyer plusieurs données asynchrones; c'est ce que fait l'objet observable et est une différence fondamentale entre les deux.

   Cela signifie que lorsque nous utilisons Rx.Observable.fromPromise (), nous obtenons un objet observable qui émet une seule valeur - ou:

   1. La valeur analysée à promettre;
   Promesse la valeur rejetée.
   Lorsque la promesse renvoie une cha?ne ou un nombre, nous n'avons rien à faire de spécial. Cependant, lorsqu'il renvoie un tableau (qui est ce qu'il est dans notre cas), nous préférons créer un objet observable qui émet le contenu du tableau plut?t que le tableau lui-même en une seule valeur.
   6.

   Ce processus est appelé aplatissement, que FlatMap () traite. Il a beaucoup de surcharges, mais nous n'utilisons que les surcharges les plus simples et les plus couramment utilisées.

   Lorsque vous utilisez FlatMap (), nous:

   Appelez FlatMap () sur les objets observables qui émettent une résolution ou un rejet à valeur unique de la promesse;
   Passez une fonction pour créer un nouvel objet observable.
   2. Dans notre exemple, nous passons rx.observable.from (), qui crée une séquence à partir des valeurs du tableau:

   Cela couvre le code dans notre courte préface:

   const task_stream =
     // 創(chuàng)建所有數(shù)據(jù)庫中任務的流
     getTasks().
       // 只獲取此用戶的任務
       filter((task) => task.user_id == user_id).
       // 獲取未完成的任務
       filter((task) => !task.completed).
       // 只獲取任務名稱
       map((task) => task.name)
   
   /* 任務如下所示：
      task = {
       user_id   : number,
       completed : boolean,
       name      : string
      }
    */

   RXJS fournit également un alias pour FlatMap (): selectMany ().

   const task_stream =
     parameter_stream.
       debounce(1000).
       map((parameter) => {
         getTasks().
           retry(3).
           filter((task) => task.user_id === user_id).
           filter((task) => task.completed === parameter).
           map((task)    => task.name)
       }).
       flatMap(Rx.Observable.from).
       distinctUntilChanged().
       update()

   combinant plusieurs flux
   Habituellement, nous aurons plusieurs flux qui doivent être combinés. Il existe de nombreuses fa?ons de combiner des flux, mais certains apparaissent plus fréquemment que d'autres.
   7.

   La connexion et la fusion sont les deux fa?ons les plus courantes de combiner des flux.

   La connexion

   crée un nouveau flux en émettant la valeur du premier flux jusqu'à ce qu'il soit terminé, puis en émettant la valeur du deuxième flux.

   Merge crée de nouveaux flux à partir de plusieurs flux en émettant la valeur de tout flux actif

   Pensez à parler à deux personnes en même temps sur Facebook Messenger. CONCAT () est une situation où vous recevez un message des deux parties mais terminez une conversation avec une personne avant de répondre à une autre personne. Merge (), c'est comme créer un chat de groupe et recevoir deux flux de messages en même temps.

   Le flux

   CONCAT () imprimera d'abord toutes les valeurs de Source1 et commencera uniquement à imprimer la valeur de Source2 après Source1 terminée.
   Merge () Le flux imprimera les valeurs de Source1 et Source2 en fonction de l'ordre re?u: il n'attend pas que le premier flux se termine avant d'émettre la valeur du deuxième flux.

   source.
     map((user) => user.website)

   8.

   Habituellement, nous voulons écouter des objets observables qui émettent des objets observables, mais nous concentrons uniquement sur les dernières émissions de la source.
   Pour étendre davantage l'analogie de Facebook Messenger, Switch () est vous… eh bien, basculez la personne que vous répondez en fonction de qui envoie actuellement le message.
   à cette fin, RXJS fournit un commutateur.
   L'interface utilisateur fournit plusieurs bons cas d'utilisation pour Switch (). Si notre application fait une demande chaque fois que l'utilisateur sélectionne ce qu'il souhaite rechercher, nous pouvons supposer qu'il veut simplement voir les résultats de la dernière sélection. Par conséquent, nous utilisons Switch () pour écouter uniquement les derniers résultats de sélection.

   Soit dit en passant, nous devons nous assurer de ne pas gaspiller la bande passante et de sélectionner uniquement l'accès au serveur pour la dernière fois qu'un utilisateur fait chaque seconde. La fonction que nous utilisons pour ceci s'appelle Debounce ()

   Si vous voulez aller dans l'autre sens et suivre uniquement le premier choix, vous pouvez utiliser Throttle (). Il a la même API, mais se comporte le contraire.

   9.

   Et si nous voulons permettre aux utilisateurs de rechercher des publications ou des utilisateurs avec un ID spécifique?
   Pour la démonstration, nous créerons un autre menu déroulant et permettrons aux utilisateurs de sélectionner l'ID de l'élément qu'ils souhaitent récupérer.
   Il y a deux situations. Lorsque l'utilisateur:
   changer tout choix;
   1. Modifiez deux options.
   Répondez à tout niveau de modifications en utilisant combinelatest ()

   Dans le premier cas, nous devons créer un flux qui démarre une demande de réseau en utilisant les éléments suivants:

   La sélection récente par l'utilisateur de points de terminaison;
   L'ID récemment sélectionné par l'utilisateur.
   2. … et faites-le lorsque l'utilisateur met à jour toute sélection.
   C'est ce que est combinatest () pour:

   Chaque fois qu'un flux émet une valeur, CombineLaSest () prend la valeur émise et la associe avec le dernier élément émis par les autres flux et émet la paire en tant que tableau.

   const task_stream =
     // 創(chuàng)建所有數(shù)據(jù)庫中任務的流
     getTasks().
       // 只獲取此用戶的任務
       filter((task) => task.user_id == user_id).
       // 獲取未完成的任務
       filter((task) => !task.completed).
       // 只獲取任務名稱
       map((task) => task.name)
   
   /* 任務如下所示：
      task = {
       user_id   : number,
       completed : boolean,
       name      : string
      }
    */

   C'est plus facile à visualiser dans le graphique:

   Utilisez ZIP pour répondre uniquement aux changements dans deux flux

   const task_stream =
     parameter_stream.
       debounce(1000).
       map((parameter) => {
         getTasks().
           retry(3).
           filter((task) => task.user_id === user_id).
           filter((task) => task.completed === parameter).
           map((task)    => task.name)
       }).
       flatMap(Rx.Observable.from).
       distinctUntilChanged().
       update()

   Alors attendez que l'utilisateur met à jour sa sélection de champs ID et Point de terminaison, remplacez CombineLaSest () par zip ().
   Encore une fois, c'est plus facile à comprendre dans le graphique:

   Contrairement à Combinelatest (), Zip () attendra que les deux objets observables émettent un nouveau contenu avant d'envoyer un tableau de leurs valeurs mises à jour.

   source.
     map((user) => user.website)

   10.

   Enfin, Takeuntil () nous permet d'écouter le premier flux jusqu'à ce que le deuxième flux commence à émettre des valeurs.

   Ceci est utile lorsque vous avez besoin de coordonner les flux mais que vous n'avez pas besoin de les combiner.
   Résumé

   source.
     map((user) => user.website).
     filter((website) => (website.endsWith('net') || website.endsWith('org'));
   })

   Il suffit d'ajouter des dimensions de temps aux tableaux ouvre la porte à une nouvelle réflexion sur les programmes.

   RXJS est bien plus que ce que nous voyons ici, mais cela suffit pour aller très loin.

   Commencez par RXJS Lite, soyez prêt à vous référer à la documentation et à prendre le temps de le faire. Avant de le savoir, tout ressemblera à un flux… parce que tout est.

   FAQ sur les fonctions RXJS (FAQ)

   Quelle est la principale différence entre RXJ et JavaScript traditionnel?

   RXJS est une bibliothèque de programmation réactive utilisant des objets observables pour simplifier la combinaison du code asynchrone ou basé sur un rappel. Ceci est comparé à l'utilisation de JavaScript traditionnel avec un style de programmation plus impératif. La principale différence est la fa?on dont ils traitent les données - RXJS traite les données comme un flux, qui peut être utilisé et transformé à l'aide de divers opérateurs, tandis que JavaScript traditionnel traite les données de manière plus linéaire.

   Comment créer des objets observables dans RXJS?

   Dans RXJS, vous pouvez créer des objets observables à l'aide du nouveau constructeur observable (). Ce constructeur prend une fonction comme un argument, appelé une fonction d'abonnée, qui est exécutée lorsqu'elle est initialement abonnée à un objet observable. Voici un exemple de base:

   const task_stream =
     // 創(chuàng)建所有數(shù)據(jù)庫中任務的流
     getTasks().
       // 只獲取此用戶的任務
       filter((task) => task.user_id == user_id).
       // 獲取未完成的任務
       filter((task) => !task.completed).
       // 只獲取任務名稱
       map((task) => task.name)
   
   /* 任務如下所示：
      task = {
       user_id   : number,
       completed : boolean,
       name      : string
      }
    */

   Quels sont les principaux opérateurs de RXJ et comment ils fonctionnent?

   RXJS possède une large gamme d'opérateurs qui peuvent être utilisés pour contr?ler la fa?on dont les données s'écoulent entre les objets observables et les observateurs. Certains des principaux opérateurs incluent MAP (), Filter (), Réduction (), Merge () et Concat (). Chacun de ces opérateurs exploite un flux de données de différentes manières, tels que la conversion de données, le filtrage de certaines valeurs ou la combinaison de plusieurs flux.

   Comment gérer les erreurs dans RXJS?

   RXJS fournit plusieurs opérateurs qui gèrent les erreurs, telles que capturerror (), retRy () et retRyWhen (). L'opérateur CatchError () est utilisé pour capter des erreurs sur les flux observables et pour renvoyer un nouvel objet observable ou lancer une erreur. L'opérateur de retry () peut être utilisé pour soulager les objets observables en cas d'erreur. L'opérateur RETRYWHEN () est similaire, mais il fournit plus de contr?le sur le moment de réessayer.

   Comment annuler l'abonnement d'objets observables dans RXJS?

   Lorsque vous vous abonnez à un observable, vous recevez un abonnement qui a une méthode de désinscription (). Vous pouvez appeler cette méthode pour annuler l'exécution de l'objet observable et nettoyer toutes les ressources utilisées. Voici un exemple:

   const task_stream =
     parameter_stream.
       debounce(1000).
       map((parameter) => {
         getTasks().
           retry(3).
           filter((task) => task.user_id === user_id).
           filter((task) => task.completed === parameter).
           map((task)    => task.name)
       }).
       flatMap(Rx.Observable.from).
       distinctUntilChanged().
       update()

   Quelle est la différence entre un objet observable à chaleur moyenne et un objet observable froid?

   Dans RXJS, l'objet observable peut être chaud ou froid. Les observables à froid commencent à fonctionner lorsqu'ils sont abonnés, tandis que les observables chauds produisent des valeurs avant même de s'abonner. En d'autres termes, les objets observables à froid sont inertes, tandis que les objets observables chauds ne le sont pas.

   Comment combiner plusieurs objets observables dans RXJS?

   RXJS fournit plusieurs opérateurs qui combinent plusieurs objets observables, tels que Merge (), Concat (), CombinelaTest () et Zip (). Chacun de ces opérateurs combine des flux de données de différentes manières, selon vos besoins spécifiques.

   Quel est le but du thème dans RXJS?

   Le sujet dans RXJS est un type spécial d'objet observable qui permet la multidiffusion de valeurs à plusieurs observateurs. Contrairement aux objets observables ordinaires, les sujets maintiennent des registres pour de nombreux auditeurs.

   Comment utiliser RXJS avec Angular?

   Angular prend en charge les RXJ dans construits et l'utilise en interne pour diverses fonctions. Vous pouvez également utiliser les RXJ dans votre propre code pour gérer les opérations asynchrones et implémenter des fonctions telles que l'achèvement automatique, la déjITTER, la limitation, le sondage, etc.

   Quels sont les cas d'utilisation courants pour les RXJ?

   RXJS peut être utilisé dans divers scénarios où des données asynchrones sont nécessaires. Certains cas d'utilisation courants incluent la gestion des entrées utilisateur, la fabrication de demandes HTTP, l'utilisation de WebSockets et la gestion des animations.

   Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!