Node.js
Korzystanie ze strumieni

Parametry

Czytaj dane z pliku tekstowego za pomocą strumieni

We / wy w węźle jest asynchroniczne, więc interakcja z dyskiem i siecią wymaga przekazywania wywołań zwrotnych do funkcji. Możesz mieć ochotę napisać kod obsługujący plik z dysku w następujący sposób:

var http = require('http');
var fs = require('fs');

var server = http.createServer(function (req, res) {
    fs.readFile(__dirname + '/data.txt', function (err, data) {
        res.end(data);
    });
});
server.listen(8000);

Ten kod działa, ale jest nieporęczny i buforuje cały plik data.txt do pamięci dla każdego żądania przed zapisaniem wyniku z powrotem do klientów. Jeśli data.txt jest bardzo duży, Twój program może zacząć jeść dużo pamięci, ponieważ obsługuje wielu użytkowników jednocześnie, szczególnie dla użytkowników korzystających z wolnych połączeń.

Doświadczenie użytkownika jest również słabe, ponieważ użytkownicy będą musieli poczekać, aż cały plik zostanie buforowany w pamięci na serwerze, zanim będą mogli zacząć odbierać jakąkolwiek zawartość.

Na szczęście oba argumenty (req, res) są strumieniami, co oznacza, że możemy napisać to w znacznie lepszy sposób za pomocą fs.createReadStream () zamiast fs.readFile ():

var http = require('http');
var fs = require('fs');

var server = http.createServer(function (req, res) {
    var stream = fs.createReadStream(__dirname + '/data.txt');
    stream.pipe(res);
});
server.listen(8000);

Tutaj .pipe () zajmuje się nasłuchiwaniem zdarzeń „data” i „end” z fs.createReadStream (). Ten kod jest nie tylko bardziej przejrzysty, ale teraz plik data.txt zostanie zapisany do klientów po jednej porcji natychmiast po ich otrzymaniu z dysku.

Rurociągi

Strumienie odczytywalne można „potokować” lub połączyć z zapisywalnymi strumieniami. Dzięki temu przepływ danych ze strumienia źródłowego do strumienia docelowego jest bez większego wysiłku.

var fs = require('fs')

var readable = fs.createReadStream('file1.txt')
var writable = fs.createWriteStream('file2.txt')

readable.pipe(writable) // returns writable

Gdy zapisywalne strumienie są również strumieniami czytelnymi, tj. Gdy są strumieniami dupleksowymi , możesz kontynuować przesyłanie strumieniowe do innych zapisywalnych strumieni.

var zlib = require('zlib')

fs.createReadStream('style.css')
  .pipe(zlib.createGzip()) // The returned object, zlib.Gzip, is a duplex stream.
  .pipe(fs.createWriteStream('style.css.gz')

Czytelne strumienie można również potokować do wielu strumieni.

var readable = fs.createReadStream('source.css')
readable.pipe(zlib.createGzip()).pipe(fs.createWriteStream('output.css.gz'))
readable.pipe(fs.createWriteStream('output.css')

Pamiętaj, że musisz potokować do strumieni wyjściowych synchronicznie (w tym samym czasie), zanim jakiekolwiek dane „przepłyną”. Nieprzestrzeganie tego może prowadzić do przesyłania niekompletnych danych.

Należy również pamiętać, że obiekty strumieniowe mogą emitować zdarzenia error ; pamiętaj, aby w razie potrzeby odpowiedzialnie obsługiwać te zdarzenia w każdym strumieniu:

var readable = fs.createReadStream('file3.txt')
var writable = fs.createWriteStream('file4.txt')
readable.pipe(writable)
readable.on('error', console.error)
writable.on('error', console.error)

Tworzenie własnego strumienia do odczytu / zapisu

Zobaczymy obiekty strumieniowe zwracane przez moduły takie jak fs itp. Ale co, jeśli chcemy stworzyć własny obiekt strumieniowy.

Aby utworzyć obiekt Stream, musimy użyć modułu strumienia dostarczonego przez NodeJs

    var fs = require("fs");
    var stream = require("stream").Writable;
    
    /* 
     *  Implementing the write function in writable stream class.
     *  This is the function which will be used when other stream is piped into this 
     *  writable stream.
     */
    stream.prototype._write = function(chunk, data){
        console.log(data);
    }
    
    var customStream = new stream();
    
    fs.createReadStream("am1.js").pipe(customStream);

To da nam własny niestandardowy strumień do zapisu. możemy zaimplementować wszystko w ramach funkcji _write . Powyższa metoda działa w wersji NodeJs 4.xx, ale w NodeJs 6.x ES6 wprowadzono klasy, dlatego składnia uległa zmianie. Poniżej znajduje się kod wersji 6.x NodeJs

    const Writable = require('stream').Writable;
    
    class MyWritable extends Writable {
      constructor(options) {
        super(options);
      }
    
      _write(chunk, encoding, callback) {
        console.log(chunk);
      }
    }

Dlaczego strumienie?

Przeanalizujmy dwa następujące przykłady odczytu zawartości pliku:

Pierwszy, który wykorzystuje metodę asynchroniczną do odczytu pliku i udostępnia funkcję zwrotną, która jest wywoływana po pełnym wczytaniu pliku do pamięci:

fs.readFile(`${__dirname}/utils.js`, (err, data) => {
  if (err) {
    handleError(err);
  } else {
    console.log(data.toString());
  }
})

A drugi, który używa streams w celu odczytania zawartości pliku, kawałek po kawałku:

var fileStream = fs.createReadStream(`${__dirname}/file`);
var fileContent = '';
fileStream.on('data', data => {
  fileContent += data.toString();
})

fileStream.on('end', () => {
  console.log(fileContent);
})

fileStream.on('error', err => {
  handleError(err)
})

Warto wspomnieć, że oba przykłady robią dokładnie to samo . Jaka jest zatem różnica?

• Pierwszy jest krótszy i wygląda bardziej elegancko
• Drugi pozwala wykonać przetwarzanie pliku podczas jego odczytywania (!)

Gdy pliki, z którymi masz do czynienia, są małe, nie ma rzeczywistego efektu podczas korzystania ze streams , ale co się stanie, gdy plik jest duży? (tak duży, że odczytanie go do pamięci zajmuje 10 sekund)

Bez streams będziesz czekał, nie robiąc absolutnie nic (chyba że twój proces wykona inne czynności), aż minie 10 sekund i plik zostanie w pełni odczytany , i dopiero wtedy możesz rozpocząć przetwarzanie pliku.

Dzięki streams zawartość pliku jest przekazywana kawałek po kawałku, dokładnie wtedy, gdy są one dostępne - a to pozwala przetwarzać plik podczas jego odczytywania.

Powyższy przykład nie ilustruje, w jaki sposób można wykorzystać streams do pracy, której nie można wykonać podczas wykonywania funkcji wywołania zwrotnego, więc spójrzmy na inny przykład:

Chciałbym pobrać plik gzip , rozpakować go i zapisać jego zawartość na dysku. Biorąc pod uwagę url pliku, należy to zrobić:

• Pobierz plik
• Rozpakuj plik
• Zapisz go na dysku

Oto [mały plik] [1], który jest przechowywany w mojej pamięci S3 . Poniższy kod wykonuje powyższe czynności w trybie wywołania zwrotnego.

var startTime = Date.now()
s3.getObject({Bucket: 'some-bucket', Key: 'tweets.gz'}, (err, data) => {
  // here, the whole file was downloaded

  zlib.gunzip(data.Body, (err, data) => {
    // here, the whole file was unzipped

    fs.writeFile(`${__dirname}/tweets.json`, data, err => {
      if (err) console.error(err)

      // here, the whole file was written to disk
      var endTime = Date.now()
      console.log(`${endTime - startTime} milliseconds`) // 1339 milliseconds
    })
  })
})

// 1339 milliseconds

Tak to wygląda przy użyciu streams :

s3.getObject({Bucket: 'some-bucket', Key: 'tweets.gz'}).createReadStream()
  .pipe(zlib.createGunzip())
  .pipe(fs.createWriteStream(`${__dirname}/tweets.json`));

// 1204 milliseconds

Tak, w przypadku małych plików nie jest to szybsze - testowany plik waży 80KB . Testowanie to na większą pliku 71MB skompresowane ( 382MB rozpakowane), pokazuje, że streams wersja jest znacznie szybszy

• Zajęło 20925 milisekund pobrać 71MB , rozpakuj go, a następnie zapisu 382MB na dysku - używając modę oddzwonienia.
• Dla porównania to samo zajęło 13434 milisekundy w przypadku korzystania z wersji streams (35% szybciej, w przypadku niezbyt dużego pliku)