Einführung in Stream in Node.js_node.js

Was ist Flow?

Apropos Streams: Es handelt sich um ein *nix-Konzept: Pipe – In *nix werden Streams in der Shell als Daten implementiert, die über | (Pipe-Zeichen) überbrückt werden können. Die Ausgabe von a Der Prozess (stdout) kann direkt als Eingabe (stdin) des nächsten Prozesses verwendet werden.

In Node ist das Konzept des Streams (Stream) ähnlich und stellt die Fähigkeit eines Datenstroms dar, überbrückt zu werden.

Rohr

Die Essenz des Streamings liegt in der .pipe()-Methode. Die Fähigkeit zur Überbrückung besteht darin, dass beide Enden des Datenstroms (Upstream/Downstream oder Lese-/Schreibstrom) mit einer .pipe()-Methode überbrückt werden.

Die Ausdrucksform des Pseudocodes ist:

Code kopieren Der Code lautet wie folgt:

//upstream.pipe (downstream)
Readable.pipe(Writable);

Klassifizierung von Streams

Dies ist nicht dazu gedacht, den sogenannten „klassischen“ Ablauf vor Node v0.4 zu diskutieren. Anschließend werden Streams in mehrere Kategorien unterteilt (alle abstrakten Schnittstellen:

).

1.stream.Readable Lesbarer Stream (muss die _read-Methode implementieren, der Schwerpunkt liegt auf den Details des Lesens des Datenstroms
2.stream.Writable Beschreibbarer Stream (muss die _write-Methode implementieren, der Schwerpunkt liegt auf den Details des Schreibens des Datenstroms
3.stream.Duplex Lese-/Schreibstream (muss die beiden oben genannten Schnittstellen implementieren, konzentrieren Sie sich auf die Details der beiden oben genannten Schnittstellen
4.stream.Transform Von Duplex geerbt (muss die _transform-Methode implementieren, der Schwerpunkt liegt auf der Verarbeitung von Datenblöcken

Kurz gesagt:

1) Der Besitzer von .pipe() muss über die Fähigkeit „Lesbarer Stream“ (aber nicht darauf beschränkt) verfügen. Es verfügt über eine Reihe von „lesbaren“/„Daten“/„Ende“/„Schließen“/„Fehler“-Ereignissen Das Abonnement bietet auch eine Reihe von Methoden wie .read()/.pause()/.resume() zum Aufrufen; 2) Die Parameter von .pipe() müssen beschreibbare Stream-Funktionen haben (aber nicht beschränkt auf). Es verfügt über „drain“/„pipe“/„unpipe“/„error“/„finish“-Ereignisse für den Zugriff und bietet auch . Für den Aufruf von
stehen write()/.end() und andere Methoden zur Verfügung

Was zum Teufel

Fühlst du dich ein wenig ängstlich? Machen Sie sich keine Sorgen, als Low-Level-Programmierer, der die menschliche Sprache spricht, werde ich Stream auseinandernehmen und mit Ihnen darüber sprechen.

Die

Stream-Klasse im

Node.js-Quellcode ist wie folgt definiert:

Code kopieren Der Code lautet wie folgt:

var EE = require('events').EventEmitter;
var util = require('util');
util.inherits(Stream, EE);

Funktion Stream() {
EE.call(this);
}

Wie Sie sehen können, ist Stream im Wesentlichen ein EventEmitter, was bedeutet, dass es über ereignisgesteuerte Funktionen verfügt (.emit/.on...). Wie wir alle wissen, ist „Node.js eine ereignisgesteuerte Plattform basierend auf V8“, die ereignisgesteuerte Streaming-Programmierung implementiert und die gleichen asynchronen Rückrufeigenschaften wie Node aufweist.

Zum Beispiel gibt es in einem lesbaren Stream ein lesbares Ereignis. In einem pausierten schreibgeschützten Stream wird ein Datenblock an den Abonnenten gesendet (was sind die lesbaren Streams). ? Express) req, req.part der FTP- oder Multi-Form-Upload-Komponente, Standard-Eingabeprozess.stdin im System usw.). Mit dem lesbaren Ereignis können wir ein Tool wie einen Parser erstellen, der die Shell-Befehlsausgabe verarbeitet:

Code kopieren Der Code lautet wie folgt:

process.stdin.on('readable', function(){
var buf = process.stdin.read();
if(buf){
var data = buf.toString();
// Daten analysieren ...                                                  }
});

Appelez comme ceci :

Copier le code Le code est le suivant :

head -10 some.txt | nœud analyseur.js

Pour un flux lisible, nous pouvons également nous abonner à ses données et mettre fin aux événements pour obtenir des morceaux de données et être averti lorsque le flux est épuisé, comme dans l'exemple de socket classique :

Copier le code Le code est le suivant :

req.on('connect', function(res, socket, head) {
socket.on('data', function(morceau) {
console.log(chunk.toString());
});
socket.on('end', function() {
       proxy.close();
});
});

Changement d'état du flux lisible
Il est à noter que le flux Readable a deux états : le mode flux (torrent) et le mode pause (pause). Le premier ne peut pas s'arrêter du tout et continuera à nourrir celui qui est redirigé ; le second fera une pause jusqu'à ce que l'aval appelle explicitement la requête Stream.read() pour lire le bloc de données. Le flux Readable est en mode pause lors de son initialisation.

Ces deux états peuvent être commutés entre eux, parmi lesquels,

Si l'un des comportements suivants se produit, la pause deviendra fluide :

1. Ajoutez un abonnement à un événement de données au flux Lisible
2. Appelez .resume() sur Readable pour activer explicitement le flux
3. Appelez .pipe(writable) du flux Readable pour établir un pont vers un flux Writable

Si l'un des comportements suivants se produit, le flux reviendra en pause :

1.Le flux lisible n'a encore été redirigé vers aucun flux, .pause() réglable peut être utilisé pour mettre en pause
2. Le flux Readable a été redirigé vers le flux. Vous devez supprimer tous les abonnements aux événements de données et appeler la méthode .unpipe() pour libérer la relation avec le flux en aval un par un

Utilisation merveilleuse

Combiné aux caractéristiques asynchrones du flux, je peux écrire une application comme celle-ci : relier directement la sortie de l'utilisateur A à la sortie sur la page de l'utilisateur B :

Copier le code Le code est le suivant :

router.post('/post', function(req, res) {
var destination = req.headers['destination']; //À qui envoyer
cache[destination] = req;
//Oui, il ne revient pas, il est donc préférable de faire une requête ajax
});

Lorsque l'utilisateur B demande :

Copier le code Le code est le suivant :

router.get('/inbox', function(req, res){
var utilisateur = req.headers['user'];
cache.find(user, function(err, previousReq){ //Rechercher la demande précédemment enregistrée
      var form = new multiparty.Form();
        form.parse(previousReq); // Il y a des fichiers pour moi
       form.on('part', function (partie) {
              part.pipe(res); //La méthode de streaming est bonne :)

              part.on('erreur', fonction (erreur) {
console.log(err);
                 messagerie.setRequestDone(uniqueID);
                     return res.end(err);
            });
        });
});
});

Référence

comment écrire des programmes de nœuds avec des flux : stream-handbook