Pengenalan kepada Strim dalam Node.js_node.js

Apakah aliran?

Bercakap tentang aliran, ia melibatkan konsep *nix: Paip - Dalam *nix, aliran dilaksanakan dalam Shell sebagai data yang boleh dirapatkan melalui | proses (stdout) boleh terus digunakan sebagai input (stdin) proses seterusnya.

Dalam Node, konsep strim (Strim) adalah serupa, mewakili keupayaan strim data untuk dirapatkan.

paip

Intipati penstriman terletak pada kaedah .pipe(). Keupayaan untuk merapatkan ialah kedua-dua hujung aliran data (hulu/hiliran atau aliran baca/tulis) dirapatkan dengan kaedah .pipe().

Bentuk ungkapan kod pseudo ialah:

Salin kod Kod adalah seperti berikut:

//upstream.pipe (hiliran)
Readable.pipe(Boleh Tulis);

Klasifikasi strim

Ini tidak bertujuan untuk membincangkan apa yang dipanggil aliran "klasik" sebelum Node v0.4. Kemudian, strim dibahagikan kepada beberapa kategori (semua antara muka abstrak:

1.strim.Boleh dibaca Strim boleh dibaca (perlu melaksanakan kaedah _baca, tumpuan adalah pada butiran membaca aliran data
2.strim.Boleh Ditulis Strim Boleh Ditulis (perlu melaksanakan kaedah _write, tumpuan adalah pada butiran menulis aliran data
3.strim.Dupleks Strim baca/tulis (perlu melaksanakan dua antara muka di atas, fokus pada butiran dua antara muka di atas
4.strim.Transform Diwarisi daripada Dupleks (perlu melaksanakan kaedah _transform, tumpuan adalah pada pemprosesan blok data

Ringkasnya:

1) Pemilik .pipe() mesti mempunyai keupayaan Strim boleh dibaca (tetapi tidak terhad kepada) Ia mempunyai satu siri peristiwa 'boleh dibaca'/'data'/'end'/'close'/'error' untuk. Langganan juga menyediakan satu siri kaedah seperti .read()/.pause()/.resume() untuk memanggil
2) Parameter .pipe() mesti mempunyai keupayaan strim Boleh Tulis (tetapi tidak terhad kepada peristiwa 'drain'/'pipe'/'unpipe'/'error'/'finish' untuk akses, dan juga menyediakan . write ()/.end() dan kaedah lain tersedia untuk memanggil

Apa kejadahnya

Adakah anda berasa sedikit cemas? Jangan risau, sebagai pengekod peringkat rendah yang bercakap bahasa manusia, saya akan memecahkan Stream dan bercakap dengan anda mengenainya.

Kelas

Stream, dalam Kod sumber Node.js , ditakrifkan seperti berikut:

Salin kod Kod adalah seperti berikut:

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

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

Seperti yang anda lihat, pada asasnya, Strim ialah Pemancar Acara, yang bermaksud ia mempunyai fungsi dipacu peristiwa (.emit/.on...). Seperti yang kita sedia maklum, "Node.js ialah platform dipacu peristiwa berdasarkan V8", yang melaksanakan pengaturcaraan penstriman dipacu peristiwa dan mempunyai ciri panggil balik tak segerak yang sama seperti Node.

Sebagai contoh, dalam strim Boleh Dibaca, terdapat acara boleh dibaca Dalam strim baca sahaja yang dijeda, selagi blok data sedia untuk dibaca, ia akan dihantar kepada pelanggan (apa itu strim Boleh Dibaca. ? Express) req, req.bahagian daripada komponen muat naik ftp atau mutli-form, proses input standard.stdin dalam sistem, dsb.). Dengan acara yang boleh dibaca, kita boleh membuat alat seperti parser yang memproses output arahan shell:

Salin kod Kod adalah seperti berikut:

process.stdin.on('boleh dibaca', function(){
var buf = process.stdin.read();
jika(buf){
var data = buf.toString();
// menghuraikan data ...                                                   }
});

Rufen Sie so an:

Code kopieren Der Code lautet wie folgt:

head -10 some.txt |. node parser.js

Für einen lesbaren Stream können wir auch seine Daten und Endereignisse abonnieren, um Datenblöcke zu erhalten und benachrichtigt zu werden, wenn der Stream erschöpft ist, wie im klassischen Socket-Beispiel:

Code kopieren Der Code lautet wie folgt:

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

Lesbare Stream-Statusumschaltung
Es ist zu beachten, dass der lesbare Stream zwei Zustände hat: Fließmodus (Torrent) und Pausenmodus (Pause). Ersteres kann überhaupt nicht anhalten und füttert weiterhin jeden, der weitergeleitet wird; letzteres hält an, bis der Downstream explizit die Stream.read()-Anfrage aufruft, um den Datenblock zu lesen. Der Readable-Stream befindet sich bei der Initialisierung im Pausenmodus.

Diese beiden Zustände können untereinander umgeschaltet werden, unter anderem

Wenn eines der folgenden Verhaltensweisen auftritt, ändert sich die Pause in fließend:

1. Fügen Sie dem Readable-Stream ein Datenereignisabonnement hinzu
2. Rufen Sie .resume() für Readable auf, um den Fluss explizit zu aktivieren
3. Rufen Sie .pipe(writable) des Readable-Streams auf, um eine Brücke zu einem Writable-Stream zu schlagen

Wenn eines der folgenden Verhaltensweisen auftritt, kehrt der Fluss in die Pause zurück:

1.Der lesbare Stream wurde noch nicht an einen Stream weitergeleitet. Mit der einstellbaren Funktion .pause() kann
angehalten werden 2. Der lesbare Stream wurde an den Stream weitergeleitet. Sie müssen alle Datenereignisabonnements entfernen und die Methode .unpipe() aufrufen, um die Beziehung zum Downstream-Stream nacheinander freizugeben

Wunderbare Verwendung

In Kombination mit den asynchronen Eigenschaften des Streams kann ich eine Anwendung wie diese schreiben: Die Ausgabe von Benutzer A direkt mit der Ausgabe auf der Seite von Benutzer B verbinden:

Code kopieren Der Code lautet wie folgt:

router.post('/post', function(req, res) {
var destination = req.headers['destination']; //An wen
gesendet werden soll Cache[Ziel] = req;
//Ja, es wird nicht zurückgegeben, daher ist es am besten, eine Ajax-Anfrage zu stellen
});

Wenn Benutzer B anfordert:

Code kopieren Der Code lautet wie folgt:

router.get('/inbox', function(req, res){
var user = req.headers['user'];
Cache.find(user, function(err, previousReq){ //Finden Sie die zuvor gespeicherte Anforderung
      var form = new multiparty.Form();
        form.parse( previousReq); // Es gibt Dateien für mich
       form.on('part', function (part) {
              part.pipe(res); //Streaming-Methode ist gut:)

              part.on('error', function (err) {
console.log(err);
                 Messaging.setRequestDone(uniqueID);
                     return res.end(err);
            });
        });
});
});

Referenz

wie man Knotenprogramme mit Streams schreibt: Stream-Handbuch