Qu'est-ce qu'un flux ? Une plongée approfondie dans les flux lisibles dans Node.js

_readableState : ReadableState {.
  objectMode : false, // Pour utiliser d'autres types de données à l'exception des chaînes, des tampons et des valeurs nulles, ce mode doit être activé. highWaterMark : 16384, // Limite du niveau d'eau, 1024 * 16, par défaut 16 Ko, si cette limite est dépassée , l'appel s'arrêtera _read() lit les données dans le tampon : BufferList { head: null, tail: null, length: 0 }, // Liste chaînée du tampon, utilisée pour enregistrer les données length: 0, // La taille de l'intégralité des données du flux lisible, si objectMode est égal à buffer.length pipes : [], // Enregistre toutes les files d'attente de tuyaux qui surveillent le flux lisible : null, // L'état du flux indépendant est null, false, true
  terminé : faux, // Toutes les données ont été consommées endEmis : faux, // Si l'événement de fin a été envoyé ou non lu : faux, // Si les données sont en cours de lecture construit : vrai, // Le flux ne peut pas être traité avant il est construit ou échoue. Destroy sync: true, // S'il faut déclencher l'événement 'readable'/'data' de manière synchrone, ou attendre le prochain tick.
  needReadable: false, // S'il est nécessaire d'envoyer l'événement lisible émisReadable: false, // L'événement lisible a été envoyé readableListening: false, // S'il existe un événement d'écoute lisible curriculum vitae: false, // Si la méthode de reprise a été appelé errorEmit: false, // Erreur L'événement a été envoyé submitClose: true, // Lorsque le flux est détruit, s'il faut envoyer l'événement de fermeture autoDestroy: true, // Automatiquement détruit, il est appelé après la 'fin' l'événement est déclenché destroy: false, // Si le flux a été détruit en erreur: null, // Identifie si le flux a signalé une erreur close: false, // Si le flux a été fermé closeEmit: false, // Si la fermeture l'événement a été envoyé defaultEncoding: 'utf8', // Le format d'encodage de caractères par défaut waitDrainWriters: null, // Pointe vers la référence 'drain' Writer surveillée de l'événement, le type est null, Writable, Set<Writable>
  multiAwaitDrain: false, // S'il y a plusieurs écrivains en attente de l'événement de drain readingMore: false, // Si plus de données peuvent être lues dataEmit: false, // Les données ont été envoyées au décodeur: null, // Encodage du décodeur: null, // Encodeur[Symbole(kPause)] : null
},

si objectMode === vrai :

Les données peuvent alors être de n’importe quel type. Quelles que soient les données transmises, elles seront stockées.

modeobjet=true

 const Stream = require('stream');
const readableStream = nouveau Stream.Readable({
  modeobjet : vrai,
  lire() {},
});

readableStream.push({ nom : 'lisa'});
console.log(readableStream._readableState.buffer.tail);
readableStream.push(true);
console.log(readableStream._readableState.buffer.tail);
readableStream.push('lisa');
console.log(readableStream._readableState.buffer.tail);
readableStream.push(666);
console.log(readableStream._readableState.buffer.tail);
readableStream.push(() => {});
console.log(readableStream._readableState.buffer.tail);
readableStream.push(Symbol(1));
console.log(readableStream._readableState.buffer.tail);
readableStream.push(BigInt(123));
console.log(readableStream._readableState.buffer.tail);

Résultats en cours d'exécution :

si objectMode === false :

Ensuite, les données ne peuvent être qu'une chaîne ou un Buffer ou un Uint8Array

modeobjet=faux

 const Stream = require('stream');

const readableStream = nouveau Stream.Readable({

  ModeObjet : faux,

  lire() {},

});

readableStream.push({ nom : 'lisa'});

Résultats en cours d'exécution :

Nous créons un flux lisible dans la console via la ligne de commande du nœud pour observer les changements dans les données dans le tampon :

Bien sûr, avant de pousser des données, nous devons implémenter sa méthode _read, ou implémenter la méthode read dans les paramètres du constructeur :

 const Stream = require('stream');

const readableStream = new Stream.Readable();

RS._read = fonction (taille) {}

ou

 const Stream = require('stream');

const readableStream = nouveau Stream.Readable({
  
  lire (taille) {}

});

Après l'opération readableStream.push('abc'), le tampon actuel est :

Vous pouvez voir que les données actuelles sont stockées. Les données stockées au début et à la fin sont les codes ASCII de la chaîne « abc », et le type est de type Buffer. La longueur représente le nombre de données actuellement enregistrées plutôt que la taille de. le contenu des données.

En imprimant toutes les méthodes de BufferList, vous pouvez obtenir :

À l'exception de la jointure, qui sérialise la BufferList en une chaîne, les autres sont toutes des opérations d'accès aux données.

Je n'expliquerai pas toutes les méthodes une par une ici, mais me concentrerai sur consumer, _getString et _getBuffer.

2.2.3.1.consommer

 Adresse du code source : BufferList.consume
https://github.com/nodejs/node/blob/d5e94fa7121c9d424588f0e1a388f8c72c784622/lib/internal/streams/buffer_list.js#L80

consommer

 // Consomme une quantité spécifiée d'octets ou de caractères des données mises en mémoire tampon.
consommer(n, hasStrings) {
  const data = this.head.data;
  si (n < data.length) {
    // `slice` est le même pour les tampons et les chaînes.
    const tranche = data.slice(0, n);
    this.head.data = data.slice(n);
    tranche de retour ;
  }
  si (n === data.length) {
    // Le premier morceau correspond parfaitement.
    renvoie this.shift();
  }
  // Le résultat s'étend sur plusieurs tampons.
  return hasStrings ? this._getString(n) : this._getBuffer(n);
}

Il y a trois conditions de jugement dans le code :

• Si la longueur en octets des données consommées est inférieure à la longueur des données stockées dans le nœud principal de la liste chaînée, les n premiers octets des données du nœud principal sont pris et les données du nœud principal actuel sont définies. aux données après le découpage.

• Si les données consommées sont exactement égales à la longueur des données stockées dans le nœud principal de la liste chaînée, les données du nœud principal actuel sont renvoyées directement.

• Si la longueur des données consommées est supérieure à la longueur du nœud principal de la liste chaînée, le dernier jugement sera effectué sur la base du deuxième paramètre transmis pour déterminer si la couche inférieure de la BufferList actuelle stocke une chaîne ou un Buffer. .

2.2.3.2. _getBuffer

 Adresse du code source : BufferList._getBuffer
https://github.com/nodejs/node/blob/d5e94fa7121c9d424588f0e1a388f8c72c784622/lib/internal/streams/buffer_list.js#L137

consommer

 // Consomme une quantité spécifiée d'octets à partir des données mises en mémoire tampon.
_getBuffer(n) {
  const ret = Buffer.allocUnsafe(n);
  const retLen = n;
  soit p = this.head;
  soit c = 0 ;
  faire {
    const buf = p.data;
    si (n > buf.length) {
      TypedArrayPrototypeSet(ret, buf, retLen - n);
      n -= buf.longueur;
    } autre {
      si (n === buf.length) {
        TypedArrayPrototypeSet(ret, buf, retLen - n);
        ++c;
        si (p.suivant)
          this.head = p.next;
        autre
          this.head = this.tail = null ;
      } autre {
       TypedArrayPrototypeSet(ret,
                              nouveau Uint8Array(buf.buffer, buf.byteOffset, n),
                              retLen - n);
        this.head = p;
        p.data = buf.slice(n);
      }
      casser;
    }
    ++c;
  } while ((p = p.next) !== null);
  this.length -= c;
  retour à la retraite;
}

En général, il s'agit d'une boucle pour exploiter les nœuds de la liste chaînée et créer un nouveau tableau Buffer pour stocker les données renvoyées.

Tout d'abord, commencez à récupérer les données du nœud principal de la liste chaînée et continuez à les copier dans le tampon nouvellement créé jusqu'à ce que les données d'un certain nœud soient supérieures ou égales à la longueur à récupérer moins la longueur qui a été obtenue.

En d'autres termes, après avoir lu le dernier nœud de la liste chaînée, il n'a pas atteint la longueur souhaitée, donc le Buffer nouvellement créé est renvoyé.

2.2.3.3. _getString

 Adresse du code source : BufferList._getString
https://github.com/nodejs/node/blob/d5e94fa7121c9d424588f0e1a388f8c72c784622/lib/internal/streams/buffer_list.js#L106

consommer

 // Consomme un nombre spécifié de caractères à partir des données mises en mémoire tampon.
_getString(n) {
  soit ret = '';
  soit p = this.head;
  soit c = 0 ;
  faire {
    const str = p.data;
    si (n > str.length) {
    ret += chaîne ;
    n -= str.length;
  } autre {
    if (n === str.length) {
      ret += chaîne ;
      ++c;
      si (p.suivant)
        this.head = p.next;
      autre
        this.head = this.tail = null ;
    } autre {
      ret += StringPrototypeSlice(str, 0, n);
      this.head = p;
      p.data = StringPrototypeSlice(str, n);
    }
    casser;
    }
    ++c;
  } while ((p = p.next) !== null);
  this.length -= c;
  retour à la retraite;
}

Le fonctionnement des chaînes est le même que celui des Buffers. Il lit également les données de la tête de la liste chaînée en boucle. Il n'y a que quelques différences dans la copie et le stockage des données. L’opération _getString est de type chaîne.

2.3. Pourquoi les instances de flux lisibles d'EventEmitter ?

Pour cette question, nous devons d'abord comprendre ce qu'est le modèle de publication-abonnement. Le modèle de publication-abonnement a des applications importantes dans la plupart des API. Qu'il s'agisse de Promise ou de Redux, les API avancées basées sur le modèle de publication-abonnement peuvent être vues partout.

Son avantage est qu'il peut stocker les fonctions de rappel liées aux événements dans la file d'attente, puis demander à l'autre partie de traiter les données à un certain moment dans le futur, réalisant ainsi une séparation des préoccupations. Le producteur produit uniquement des données et informe le consommateur. , tandis que le consommateur traite ensuite uniquement les événements correspondants et leurs données correspondantes, et le modèle de streaming Node.js correspond parfaitement à cette caractéristique.

Alors, comment le flux Node.js implémente-t-il la création d'instances basées sur EventEmitter ?

 Le code source est ici : stream/legacy
https://github.com/nodejs/node/blob/d5e94fa7121c9d424588f0e1a388f8c72c784622/lib/internal/streams/legacy.js#L10

héritage

 fonction Flux (opt) {
  EE.call (ceci, opte);
}
ObjectSetPrototypeOf(Stream.prototype, EE.prototype);
ObjectSetPrototypeOf(Flux, EE);

Ensuite, il y a ces lignes de code dans le code source du flux lisible :

 Cette partie du code source est ici : lisible
https://github.com/nodejs/node/blob/d5e94fa7121c9d424588f0e1a388f8c72c784622/lib/internal/streams/readable.js#L77

héritage

 ObjectSetPrototypeOf(Readable.prototype, Stream.prototype);
ObjectSetPrototypeOf (Lisible, Flux);

Tout d’abord, héritez de l’objet prototype de Stream d’EventEmitter, afin que toutes les instances de Stream puissent accéder aux méthodes sur EventEmitter.

Dans le même temps, les méthodes statiques sur EventEmitter sont également héritées via ObjectSetPrototypeOf(Stream, EE), et dans le constructeur de Stream, le constructeur EE est emprunté pour réaliser l'héritage de toutes les propriétés dans EventEmitter, puis dans le flux lisible, utiliser la même méthode La méthode implémente l'héritage prototypique et l'héritage des propriétés statiques de la classe Stream, obtenant ainsi :

Readable.prototype.__proto__ === Stream.prototype;

Stream.prototype.__proto__ === EE.prototype

donc:

Lisible.prototype.__proto__.__proto__ === EE.prototype

Par conséquent, vous pouvez trouver le prototype d'EventEmitter en traçant la chaîne de prototypes du flux lisible et réaliser l'héritage d'EventEmitter.

2.4. Implémentation des API associées

Les API seront affichées ici dans l'ordre dans lequel elles apparaissent dans les documents de code source, et seules les implémentations principales des API seront expliquées.

Remarque : Seules les fonctions déclarées dans le code source du flux lisible Node.js sont interprétées ici, et les définitions de fonctions introduites en externe ne sont pas incluses. Afin de réduire la longueur, tous les codes ne seront pas copiés.

Prototype lisible

 Flux {
  détruire : [Fonction : détruire],
  _undestroy : [Fonction : unestroy],
  _destroy : [Fonction (anonyme)],
  push : [Fonction (anonyme)],
  unshift : [Fonction (anonyme)],
  isPaused : [Fonction (anonyme)],
  setEncoding : [Fonction (anonyme)],
  lire : [Fonction (anonyme)],
  _read : [Fonction (anonyme)],
  pipe : [Fonction (anonyme)],
  unpipe : [Fonction (anonyme)],
  sur : [Fonction (anonyme)],
  addListener : [Fonction (anonyme)],
  removeListener : [Fonction (anonyme)],
  off : [Fonction (anonyme)],
  removeAllListeners : [Fonction (anonyme)],
  CV : [Fonction (anonyme)],
  pause : [Fonction (anonyme)],
  wrap: [Fonction (anonyme)],
  itérateur : [Fonction (anonyme)],
  [Symbole(nodejs.rejection)] : [Fonction (anonyme)],
  [Symbol(Symbol.asyncIterator)] : [Fonction (anonyme)]
}

lisible.push

 Readable.prototype.push = fonction (morceau, encodage) {
  return readableAddChunk(this, chunk, encoding, false);
} ;

La fonction principale de la méthode push est de transmettre le bloc de données au pipeline en aval en déclenchant l'événement « data », ou de stocker les données dans son propre tampon.

Le code suivant est un pseudocode pertinent et montre uniquement le processus principal :

lisible.push

 function readableAddChunk (stream, chunk, encoding, addToFront) {
  état const = flux._readableState ;
  if (chunk === null) { // pousse le signal de fin de flux nul, plus aucune donnée ne peut être écrite après cet état.reading = false;
    onEofChunk (flux, état);
  } else if (!state.objectMode) { // Si ce n'est pas le mode objet if (typeof chunk === 'string') {
      morceau = Buffer.from(morceau);
    } else if (chunk instanceof Buffer) { //S'il s'agit de Buffer
    // Traitez l'encodage} else if (Stream._isUint8Array(chunk)) {
      chunk = Stream._uint8ArrayToBuffer(chunk);
    } sinon if (morceau != null) {
      err = new ERR_INVALID_ARG_TYPE('chunk', ['string', 'Buffer', 'Uint8Array'], chunk);
    }
  }

  if (state.objectMode || (chunk && chunk.length > 0)) { // C'est le mode objet ou le chunk est Buffer
    // Le jugement de plusieurs méthodes d'insertion de données est omis ici addChunk(stream, state, chunk, true);
  }
}

fonction addChunk (flux, état, morceau, addToFront) {
  if (state.flowing && state.length === 0 && !state.sync &&
    stream.listenerCount('data') > 0) { // En mode streaming, des abonnés écoutent des données stream.emit('data', chunk);
  } else { // Sinon, enregistrez les données dans le tampon state.length += state.objectMode 1 : chunk.length;
    si (addToFront) {
      state.buffer.unshift(morceau);
    } autre {
      state.buffer.push(morceau);
    }
  }
  MaybeReadMore(stream, state); // Essayez de lire un peu plus de données}

L'opération push est principalement divisée en jugeant le mode objet. Différents types effectueront différentes opérations sur les données entrantes :

• objectMode === false : convertir les données (morceaux) en tampon
• objectMode === true : transmettre les données intactes en aval

Le premier jugement de addChunk est principalement de gérer la situation où Readable est en mode fluide, dispose d'un écouteur de données et les données du tampon sont vides.

À ce stade, les données sont principalement transmises à d'autres programmes abonnés à l'événement de données, sinon les données sont enregistrées dans la mémoire tampon.

À l'exception du jugement des conditions aux limites et de l'état de l'écoulement, cette méthode comporte principalement deux opérations.

• Appelez la méthode _read implémentée par l'utilisateur pour traiter les résultats de l'exécution

• Lire les données du tampon et déclencher l'événement « données »

lisible.lu

 // Si la longueur de lecture est supérieure à hwm, hwm sera recalculé
si (n > state.highWaterMark) {
  state.highWaterMark = calculateNewHighWaterMark(n);  
}
// Appel de la méthode _read implémentée par l'utilisateur try {
  résultat const = this._read(state.highWaterMark);
  si (résultat != null) {
    const then = résultat.then;
    if (typeof then === 'fonction') {
      alors.appel(
        résultat,
        non,
        fonction (erreur) {
          errorOrDestroy(this, err);
        });
    }
  }
} attraper (erreur) {
  errorOrDestroy(this, err);
}

Si la méthode _read implémentée par l'utilisateur renvoie une promesse, appelez la méthode then de cette promesse et transmettez les rappels de réussite et d'échec pour faciliter la gestion des exceptions.

Le code principal de la méthode read pour lire les données de zone à partir du tampon est le suivant :

lisible.lu

 fonction fromList(n, état) {
  // rien n'est mis en mémoire tampon.
  si (état.longueur === 0)
    renvoie null ;
  laisser ret;
  si (état.objectMode)
    ret = state.buffer.shift();
  else if (!n || n >= state.length) { // Gère le cas où n est vide ou supérieur à la longueur du tampon // Lit tout, tronque la liste.
    if (state.decoder) // S'il existe un décodeur, sérialisez le résultat dans une chaîne ret = state.buffer.join('');
    else if (state.buffer.length === 1) // Il n'y a qu'une seule donnée, renvoie les données du nœud principal ret = state.buffer.first();
    else // Stocke toutes les données dans un Buffer ret = state.buffer.concat(state.length);
    state.buffer.clear(); // Effacer le tampon} else {
    // Gère la situation où la longueur de lecture est inférieure à la taille du tampon ret = state.buffer.consume(n, state.decoder);
  }
  retour à la retraite;
}

Une méthode qui doit être implémentée lorsque les utilisateurs initialisent un flux Readable. Vous pouvez appeler la méthode push dans cette méthode pour déclencher en continu la méthode read. Lorsque nous poussons null, nous pouvons arrêter l'opération d'écriture du flux.

Exemple de code :

lisible._read

 const Stream = require('stream');
const readableStream = nouveau Stream.Readable({
  lire (hwm) {
    this.push(String.fromCharCode(this.currentCharCode++));
    si (this.currentCharCode > 122) {
      this.push(null);
    }
  },
});
readableStream.currentCharCode = 97 ;
readableStream.pipe(process.stdout);
// abcdefghijklmnopqrstuvwxyz%

Liez un ou plusieurs flux inscriptibles au flux lisible actuel et faites passer le flux lisible en mode fluide.

Il existe de nombreux handles d’écoute d’événements dans cette méthode, et je ne les présenterai pas un par un ici :

lisible.pipe

 Readable.prototype.pipe = fonction (dest, pipeOpts) {
  const src = ceci ;
  const state = this._readableState ;
  state.pipes.push(dest); // Collecte le flux inscriptible src.on('data', ondata);
  fonction ondata (morceau) {
    const ret = dest.write(morceau);
    si (ret === faux) {
      pause();
    }
  }
  // Dites à la destination vers laquelle il est redirigé.
  dest.emit('pipe', src);
  // Démarre le flux si le flux est en mode pause if (dest.writableNeedDrain === true) {
    si (état.flux) {
      pause();
    }
  } sinon si (!state.flowing) {
    src.resume();
  }
  retourner la destination ;
}

L'opération de canal est très similaire à l'opérateur de canal Linux '|', changeant la sortie gauche en entrée droite. Cette méthode collecte le flux inscriptible pour la maintenance et déclenche l'événement 'data' lorsque le flux lisible est disponible.

Lorsque les données sortent, l'événement d'écriture du flux inscriptible sera déclenché, afin que les données puissent être transférées et que des opérations telles qu'un pipeline puissent être réalisées. Et changera automatiquement le flux lisible en mode pause en mode fluide.

Basculez le flux du mode « pause » au mode « flux ». Si l'écouteur d'événements « lisible » est défini, cette méthode n'a aucun effet.

lisible.resume

 Readable.prototype.resume = fonction() {
  const state = this._readableState ;
  si (!state.flowing) {
    state.flowing = !state.readableListening; // Le fait qu'il soit en mode fluide dépend de la définition ou non du handle d'écoute « lisible » curriculum vitae (this, state);
  }
} ;

fonction reprendre (flux, état) {
  if (!state.resumeScheduled) { // Basculez pour que la méthode curriculum vitae_ ne soit appelée qu'une seule fois dans le même Tick state.resumeScheduled = true;
    process.nextTick(resume_, stream, state);
  }
}

fonction curriculum vitae_(flux, état) {
  si (!state.reading) {
    flux.read(0);
  }
  state.resumeScheduled = faux ;
  stream.emit('resume');
  flux(flux);
}

function flow(stream) { // Lorsque le flux est en mode streaming, cette méthode continuera à lire les données du tampon jusqu'à ce que le tampon soit vide const state = stream._readableState;
  while (state.flowing && stream.read() !== null); 
  // Étant donné que la méthode read sera appelée ici et que le flux de l'écouteur d'événement 'readable' est défini, la méthode read peut également être appelée.
  // Cela entraîne des données incohérentes (n'affecte pas les données, affecte uniquement l'appel de la méthode read dans le rappel d'événement 'readable' pour lire les données)
}

Changez le flux du mode fluide au mode pause, arrêtez de déclencher l'événement « données » et enregistrez toutes les données dans le tampon.

lisible.pause

 Readable.prototype.pause = fonction() {
  if (this._readableState.flowing !== false) {
    debug('pause');
    this._readableState.flowing = false;
    this.emit('pause');
  }
  rends ceci ;
} ;

2.5. Utilisation et mécanisme de travail

La méthode d'utilisation a été mentionnée dans la section BufferList. Créez une instance Readable et implémentez sa méthode _read(), ou implémentez la méthode read dans le premier paramètre objet du constructeur.

Ici, nous dessinons uniquement le processus général et les conditions de déclenchement de la conversion de mode du flux Readable.

dans:

• needReadable(true) : mode pause et données du tampon <=hwm, la fonction d'écoute des événements lisibles est liée, il n'y a pas de données dans le tampon ou les données renvoyées sont vides lors de la lecture des données.
• push : en mode flux, il n'y a pas de données dans le tampon et l'événement 'data' sera déclenché ; sinon, les données seront enregistrées dans le tampon et l'événement 'readable' sera déclenché en fonction de l'état needReadable.
• read : lors de la lecture de données avec une longueur = 0, les données dans le tampon ont atteint hwm ou un débordement doit déclencher l'événement « lisible » ; lire les données du tampon et déclencher l'événement « data » ;
• CV : s'il existe un écouteur « lisible », cette méthode ne fonctionnera pas ; sinon, le flux passera du mode pause au mode flux et les données dans le tampon seront effacées.
• Conditions de déclenchement lisibles : l'événement « lisible » est lié et il y a des données dans le tampon. Lors du transfert de données, il y a des données dans le tampon, et needReadable === true Lors de la lecture de données avec une longueur = 0, les données dans le tampon. a atteint le hwm ou le débordement

3. Résumé

• Afin de résoudre le problème de mémoire et de temps, Node.js implémente son propre flux, afin que les données puissent être lues dans la mémoire en petits morceaux pour que les consommateurs puissent les consommer.
• Les flux ne sont pas un concept spécifique à Node.js, ils ont été introduits dans le système d'exploitation Unix il y a plusieurs décennies
• Il existe quatre types de flux : le flux lisible, le flux inscriptible, le flux lisible et inscriptible et le flux de conversion. Ils héritent tous des méthodes d'instance et des méthodes statiques d'EventEmiiter et sont toutes des instances d'EE.
• Le conteneur sous-jacent du flux est basé sur BufferList, qui est une implémentation de liste chaînée personnalisée. Chaque tête et queue est un « pointeur » pointant vers la référence de nœud suivante.
• Les flux lisibles ont deux modes et trois états. En mode fluide, les données seront envoyées aux consommateurs via EventEmitter.
• Sur la base des flux, nous pouvons implémenter un traitement en chaîne des données et assembler différentes fonctions de traitement de flux pour implémenter diverses opérations sur les flux et les convertir en données souhaitées.