Notes d'étude Nodejs Stream module_node.js-js tutoriel-php.cn

1, analyse d'ouverture

Stream est une interface abstraite implémentée par de nombreux objets dans Node. Par exemple, une requête adressée à un serveur HTTP est un flux, et stdout est également un flux. Les flux sont lisibles, inscriptibles ou les deux.

La première exposition à Stream a commencé dans les premiers jours d'Unix. Des décennies de pratique ont prouvé que l'idée de Stream peut facilement développer d'énormes systèmes.

Sous Unix, Stream est implémenté via "|". Dans Node, en tant que module de flux intégré, de nombreux modules de base et modules tiers sont utilisés.

Comme Unix, l'opération principale du flux de nœuds est également .pipe(). Les utilisateurs peuvent utiliser le mécanisme anti-pression pour contrôler l'équilibre de la lecture et de l'écriture.

Stream peut fournir aux développeurs une interface unifiée qui peut être réutilisée et contrôler l'équilibre de lecture et d'écriture entre les flux via l'interface abstraite Stream.

Une connexion TCP est à la fois un flux lisible et un flux inscriptible, tandis qu'une connexion HTTP est différente. Un objet de requête http est un flux lisible et un objet de réponse http est un flux inscriptible.

Le processus de transmission du flux est transmis sous forme de tampon par défaut, sauf si vous définissez d'autres méthodes de codage pour celui-ci. Voici un exemple :

Copier le code Le code est le suivant :

var http = require('http') ;

var serveur = http.createServer(function(req,res){

res.writeHeader(200, {'Content-Type': 'text/plain'}) ;

res.end("Bonjour Gros Ours!") ;

}) ;

serveur.écouter(8888) ;

console.log("serveur http exécuté sur le port 8888...") ;

Des caractères tronqués apparaîtront après l'exécution. La raison est que le jeu de caractères spécifié n'est pas défini, tel que : "utf-8".

Modifiez-le simplement :

Copier le code Le code est le suivant :

var http = require('http') ;

var serveur = http.createServer(function(req,res){

res.writeHeader(200,{

         'Content-Type' : 'text/plain;charset=utf-8' // Ajouter charset=utf-8

}) ;

res.end("Bonjour Gros Ours!") ;

}) ;

serveur.écouter(8888) ;

console.log("serveur http exécuté sur le port 8888...") ;

Résultat de l'exécution :

Pourquoi utiliser Stream
Les E/S dans le nœud sont asynchrones, donc la lecture et l'écriture sur le disque et le réseau nécessitent des fonctions de rappel pour lire les données. Voici un exemple de téléchargement de fichier
. Code ci-dessus :

Copier le code Le code est le suivant :

var http = require('http') ;

var fs = require('fs') ;

var serveur = http.createServer (fonction (req, res) {

fs.readFile(__dirname '/data.txt', fonction (erreur, données) {

           res.end(data);

}) ;

}) ;

serveur.écouter(8888) ;

Le code peut réaliser les fonctions requises, mais le service doit mettre en cache toutes les données du fichier en mémoire avant d'envoyer les données du fichier si le fichier "data.txt" est très volumineux
. S'il est volumineux et que la quantité de concurrence est importante, beaucoup de mémoire sera gaspillée. Étant donné que l'utilisateur doit attendre que l'intégralité du fichier soit mise en cache en mémoire avant d'accepter les données du fichier, cela entraîne
L'expérience utilisateur est assez mauvaise. Heureusement, les deux paramètres (req, res) sont Stream, nous pouvons donc utiliser fs.createReadStream() au lieu de fs.readFile(). Comme suit :

Copier le code Le code est le suivant :

var http = require('http') ;

var fs = require('fs') ;

var serveur = http.createServer (fonction (req, res) {

var stream = fs.createReadStream(__dirname '/data.txt') ;

Stream.pipe(res) ;

}) ;

serveur.écouter(8888) ;

La méthode .pipe() écoute les événements 'data' et 'end' de fs.createReadStream(), de sorte que le fichier "data.txt" n'a pas besoin d'être mis en cache dans son intégralité
fichier, un bloc de données peut être envoyé au client immédiatement après l'établissement de la connexion client. Un autre avantage de l'utilisation de .pipe() est qu'il peut résoudre le problème lorsque le client
Le problème de déséquilibre lecture-écriture causé par une très grande latence de bout en bout.

Il existe cinq flux de base : lisible, inscriptible, transformé, duplex et "classique". (Veuillez vérifier l'API pour une utilisation spécifique)

2. Introduction des exemples

Lorsque les données à traiter ne peuvent pas être chargées dans la mémoire en même temps, ou lorsqu'il est plus efficace de lire et de traiter en même temps, nous devons utiliser des flux de données. NodeJS fournit des opérations sur les flux de données via divers Streams.

En prenant le programme de copie de fichiers volumineux comme exemple, nous pouvons créer un flux de données en lecture seule pour la source de données. L'exemple est le suivant :

Copier le code Le code est le suivant :

var rs = fs.createReadStream(pathname);

rs.on('data', fonction (morceau) {

doSomething(chunk); // N'hésitez pas à jouer avec les détails

});

rs.on('end', function() {

       cleanUp() ;

}) ;

L'événement data dans le code sera déclenché en continu, que la fonction doSomething puisse ou non le gérer. Le code peut continuer à être modifié comme suit pour résoudre ce problème.

Copier le code Le code est le suivant :

var rs = fs.createReadStream(src) ;

rs.on('data', fonction (morceau) {

​ rs.pause() ;

faire quelque chose (morceau, fonction () {

         rs.resume() ;

}) ;

}) ;

rs.on('end', function() {

       cleanUp();

}) ;

Un rappel est ajouté à la fonction doSomething afin que nous puissions suspendre la lecture des données avant de traiter les données et continuer à lire les données après le traitement des données.

De plus, nous pouvons également créer un flux de données en écriture seule pour la cible de données, comme suit :

Copier le code Le code est le suivant :

var rs = fs.createReadStream(src) ;

var ws = fs.createWriteStream(dst) ;

rs.on('data', fonction (morceau) {

            ws.write(chunk);

}) ;

rs.on('end', function() {

          ws.end();

}) ;

Une fois doSomething remplacé par l'écriture de données dans un flux de données en écriture seule, le code ci-dessus ressemble à un programme de copie de fichiers. Cependant, le code ci-dessus présente le problème mentionné ci-dessus. Si la vitesse d'écriture ne peut pas suivre la vitesse de lecture, le cache à l'intérieur du flux de données en écriture seule éclatera. Nous pouvons utiliser la valeur de retour de la méthode .write pour déterminer si les données entrantes sont écrites sur la cible ou temporairement placées dans le cache, et en fonction de l'événement drain, nous pouvons déterminer quand le flux de données en écriture seule a écrit les données. dans le cache à la cible, les prochaines données à écrire peuvent être transmises. Le code est donc le suivant :

Copier le code Le code est le suivant :

var rs = fs.createReadStream(src) ;

var ws = fs.createWriteStream(dst) ;

rs.on('data', fonction (morceau) {

Si (ws.write(chunk) === false) {

          rs.pause() ;

>

}) ;

rs.on('end', function() {

          ws.end();

});

ws.on('drain', function() {

​ rs.resume();

}) ;

Enfin réalisé le transfert de données d'un flux de données en lecture seule vers un flux de données en écriture seule, et inclus un contrôle d'entrepôt antidéflagrant. Comme il existe de nombreux scénarios d'utilisation pour cela, comme le programme de copie de fichiers volumineux ci-dessus, NodeJS fournit directement la méthode .pipe pour ce faire, et son implémentation interne est similaire au code ci-dessus.

Voici un processus plus complet de copie de fichiers :

Copier le code Le code est le suivant :

var fs = require('fs'),

  chemin = require('chemin'),

  out = process.stdout;

var filePath = '/bb/bigbear.mkv';

var readStream = fs.createReadStream(filePath);

var writeStream = fs.createWriteStream('file.mkv');

var stat = fs.statSync(filePath);

var totalSize = stat.size;

var passéLongueur = 0;

var tailledernière = 0;

var startTime = Date.now();

readStream.on('data', function(morceau) {

  passéLength = chunk.length;

  if (writeStream.write(chunk) === false) {

    readStream.pause();

  >

});

readStream.on('end', function() {

  writeStream.end();

});

writeStream.on('drain', function() {

  readStream.resume();

});

setTimeout (fonction show() {

  var percent = Math.ceil((passedLength / totalSize) * 100);

  var size = Math.ceil(passedLength / 1000000);

  var diff = taille - lastSize;

  lastSize = taille;

  out.clearLine();

  out.cursorTo(0);

  out.write('已完成' size 'Mo, ' percent '%, 速度：' diff * 2 'Mo/s');

  if (passedLength < totalSize) {

    setTimeout(show, 500);

  } autre {

    var endTime = Date.now();

    console.log();

    console.log('共用时：' (endTime - startTime) / 1000 '秒。');

  >

}, 500);