Node.js
ストリームの使用

パラメーター

ストリームでTextFileからデータを読み込む

ノード内のI / Oは非同期であるため、ディスクやネットワークと対話するにはコールバックを関数に渡す必要があります。あなたは、次のようにディスクからファイルを提供するコードを書こうとするかもしれません。

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);

このコードは機能しますが、大量であり、結果をクライアントに書き戻す前に、すべての要求に対してdata.txtファイル全体をメモリにバッファします。 data.txtが非常に大きい場合は、特に遅い接続のユーザーの場合、多くのユーザーに同時にサービスを提供するため、プログラムのメモリ使用量が多くなります。

ユーザーは、コンテンツの受信を開始する前に、ファイル全体がサーバー上のメモリにバッファリングされるのを待つ必要があるため、ユーザーエクスペリエンスが低下します。

幸いなことに、（req、res）引数は両方ともストリームです。つまり、fs.readFile（）の代わりにfs.createReadStream（）を使用すると、より良い方法で記述できます。

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);

ここでは、.pipe（）はfs.createReadStream（）から 'data'イベントと 'end'イベントをリスンします。このコードはより洗練されているだけでなく、data.txtファイルはディスクから受信されると同時に一度に1つのチャンクに書き込まれます。

配管の流れ

読み取り可能なストリームは、書き込み可能なストリームに「パイプ接続」または接続できます。これにより、多くの労力を要することなく、ソースストリームからデスティネーションストリームへのデータフローが行われます。

var fs = require('fs')

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

readable.pipe(writable) // returns writable

書き込み可能なストリームが読み込み可能なストリームの場合、つまりデュプレックスストリームの場合は、他の書き込み可能なストリームに引き続き引き続き接続できます。

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')

読み込み可能なストリームは、複数のストリームにパイプすることもできます。

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

データの流れの前に同時に（同時に）出力ストリームにパイプする必要があることに注意してください。そうしないと、不完全なデータがストリーミングされる可能性があります。

また、ストリームオブジェクトはerrorイベントをerrorせることがerrorます。必要に応じて、 すべてのストリームでこれらのイベントを確実に処理してください。

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)

独自の読み書き可能なストリームを作成する

ストリームオブジェクトはfsなどのモジュールによって返されますが、独自のストリーム可能オブジェクトを作成する場合はどうなりますか。

Streamオブジェクトを作成するには、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);

これにより、独自のカスタム書き込み可能なストリームが得られます。 _write関数内には何も実装できます。上記のメソッドはNodeJs 4.xxバージョンで動作しますが、NodeJs 6.x ES6ではクラスが導入されたため、構文が変更されました。以下は、NodeJsの6.xバージョンのコードです

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

なぜストリーム？

ファイルの内容を読むための次の2つの例を調べてみましょう。

最初のものは、ファイルを読み込むための非同期メソッドを使用し、ファイルがメモリに完全に読み込まれると呼び出されるコールバック関数を提供します。

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

そして、ファイルの内容を読み込むためにstreamsを使用する第2の方法は、

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)
})

どちらの例もまったく同じことを言及する価値があります。違いは何ですか？

• 最初のものは短く、よりエレガントに見える
• 2つ目は、ファイルを読み込んでいる間にそのファイルを処理します（！）

対処するファイルが小さい場合、 streamsを使用するときに実際の効果はありませんが、ファイルが大きくなるとどうなりますか？ （非常に大きいので、それをメモリに読み込むのに10秒かかる）

streamsがなければ、10秒経過してファイルが完全に読み込まれ 、ファイルの処理が開始されるまで、あなたのプロセスが他のことをしない限り、何もしないで待っています。

streamsを使用すると、ファイルの内容が利用可能になったときにすぐに取得され、読み込み中にそのファイルを処理できるようになります。

上の例は、コールバックファッションに行くときに行えない仕事のためにstreamsをどのように利用できるかを示していないので、別の例を見てみましょう：

私はgzipファイルをダウンロードし、解凍してその内容をディスクに保存したいと思います。ファイルのurlこれは実行する必要があります：

• ファイルをダウンロードする
• ファイルを解凍する
• ディスクに保存する

ここに私のS3ストレージに保存されている[小さなファイル] [1]があります。次のコードは、コールバックの方法で上記を行います。

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

これはstreamsを使用して見える方法です：

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

// 1204 milliseconds

80KB 、小さなファイルを扱うときには高速ではありません。テストされたファイルの重量は80KBです。これをより大きなファイル（ 71MB gzip（圧縮されていない382MB ））でテストすると、 streamsバージョンがはるかに高速であることが382MBます

• これは、ダウンロードするには20925ミリ秒を要した71MB 、それを解凍してから書き込み382MB コールバックのファッションを使用して -ディスクへ。
• これとは対照的に、 streamsバージョンを使用する場合、同じ処理を行うには13434ミリ秒かかりました（それほど大きなファイルでは35％高速）