Introduction à Stream dans Node.js_node.js

Qu’est-ce que le flux ?

En parlant de flux, cela implique un concept *nix : Pipe - Dans *nix, les flux sont implémentés dans le Shell en tant que données qui peuvent être pontées via | (caractère pipe), a La sortie d'un Le processus (stdout) peut être directement utilisé comme entrée (stdin) du processus suivant.

Dans Node, le concept de flux (Stream) est similaire, représentant la capacité d'un flux de données à être ponté.

tuyau

L'essence du streaming réside dans la méthode .pipe(). La possibilité de ponter réside dans le fait que les deux extrémités du flux de données (flux amont/aval ou lecture/écriture) sont pontées avec une méthode .pipe().

La forme d'expression du pseudo-code est :

Copier le code Le code est le suivant :

//upstream.pipe (en aval)
Readable.pipe(Inscriptible);

Classification des flux

Ceci n'a pas pour but d'aborder le flux dit "classique" avant Node v0.4. Ensuite, les flux sont répartis en plusieurs catégories (toutes les interfaces abstraites :

1.stream.Readable Flux lisible (doit implémenter la méthode _read, l'accent est mis sur les détails de la lecture du flux de données
2.stream.Writable Flux inscriptible (nécessite d'implémenter la méthode _write, l'accent est mis sur les détails de l'écriture du flux de données
3.stream.Duplex Flux de lecture/écriture (doit implémenter les deux interfaces ci-dessus, concentrez-vous sur les détails des deux interfaces ci-dessus
4.stream.Transform Hérité de Duplex (nécessite d'implémenter la méthode _transform, l'accent est mis sur le traitement des blocs de données

En bref :

1) Le propriétaire de .pipe() doit avoir la capacité de flux lisible (mais sans s'y limiter). Il dispose d'une série d'événements 'readable'/'data'/'end'/'close'/'error' pour. L'abonnement fournit également une série de méthodes telles que .read()/.pause()/.resume() pour appeler
 ; 2) Les paramètres de .pipe() doivent avoir des capacités de flux inscriptibles (mais sans s'y limiter). Il a des événements 'drain'/'pipe'/'unpipe'/'error'/'finish' pour l'accès et fournit également . write ()/.end() et d'autres méthodes sont disponibles pour appeler

C'est quoi ce bordel

Vous sentez-vous un peu anxieux ? Ne vous inquiétez pas, en tant que codeur de bas niveau parlant un langage humain, je vais décomposer Stream et vous en parler.

La classe

Stream, dans le code source de Node.js , est définie comme suit :

Copier le code Le code est le suivant :

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

fonction Flux() {
EE.call(this);
>

Comme vous pouvez le voir, essentiellement, Stream est un EventEmitter, ce qui signifie qu'il possède des fonctions basées sur les événements (.emit/.on...). Comme nous le savons tous, "Node.js est une plate-forme événementielle basée sur V8", qui implémente une programmation de streaming événementielle et possède les mêmes caractéristiques de rappel asynchrone que Node.

Par exemple, dans un flux Lisible, il y a un événement lisible. Dans un flux en lecture seule en pause, tant qu'un bloc de données est prêt à être lu, il sera envoyé à l'abonné (quels sont les flux Lisibles). ? Express) req, req.part du composant de téléchargement ftp ou multi-forme, processus d'entrée standard.stdin dans le système, etc.). Avec l'événement readable, nous pouvons créer un outil tel qu'un analyseur qui traite la sortie de la commande shell :

Copier le code Le code est le suivant :

process.stdin.on('readable', function(){
var buf = process.stdin.read();
si(buf){
var data = buf.toString();
// analyse des données ...                                                   >
});

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