🍥 👩🏼 🆓 NodeJSのストリーム-2回入る川 🏿 📰 👨🏼‍🔧

ストリームは、I / O操作で1つのプログラムから別のプログラムにデータを転送するためにUNIXシステムに最初に実装された概念です。これにより、各プログラムはそれが行うことで非常に特殊化され、独立したモジュールになります。このような単純なプログラムの組み合わせは、一連の呼び出しに「組み合わせ」て、より複雑なシステムを作成するのに役立ちます。

ストリームを使用すると、小さな部分でデータを交換できるため、作業中に大量のメモリを消費することがなくなります。もちろん、これは内部フロー機能の実装方法に依存します。

一般的なタスクは、大きなファイルを解析することです。たとえば、ログデータを含むテキストファイルでは、特定のテキストを含む行を見つける必要があります。ファイルをメモリに完全に読み込んでから、必要な行を探してその行を解析し始める代わりに、少しずつ読むことができます。したがって、必要以上のメモリを占有するのではなく、読み取りデータをバッファリングするために必要なだけのメモリを占有します。必要なレコードが見つかったら、すぐにそれ以上の作業を停止します。または、見つかったレコードをチェーン内の別のストリームに転送して、たとえば別の形式に変換したり、別のファイルに保存したりできます。

ストリームモジュールは、Node.JSでストリームを操作するための基本的なAPIを提供します。 Node.JSのドキュメントはこの問題を理解するのに十分ですが、いくつかのポイントの説明を含むチートシートのようなものを作成しようとします。

ストリームの種類

スレッドには4つのタイプがあります。

読み取り可能-読み取り用のデータを提供するストリーム。
書き込み可能-データを書き込むことができるストリーム。
二重-データの読み取り（読み取り可能）と書き込み（書き込み可能）の両方が可能なストリーム。読み取りと書き込みのプロセスは互いに独立して行われます。
変換は、ストリームにデータを書き込んだりストリームからデータを読み込んだりするときにデータを変更できるデュプレックスストリームの一種です（データ転送チェーンの中間リンクとしてよく使用されます）。

EventEmitterのストリームインスタンス

すべてのストリームはEventEmitterのインスタンスです。つまり、StreamClass.emit（ 'eventName'、data）イベントを生成し、それらを処理できますStreamClass.on（ 'eventName'、（data）=> {}）;

パイプ方式

あるスレッドから別のスレッドにデータを転送するには、スレッドでパイプメソッドを呼び出す最も簡単な方法は次のとおりです。

Readable.pipe(Writable);//,  "" DataBase -> File Readable.pipe(Transform).pipe(Writable);//DataBase ->   JSON  ->  JSON  File Duplex.pipe(Transform).pipe(Duplex);//  DataBase ->  ->    DataBase

呼び出しの最後のチェーンは、それぞれが独自の問題を解決するような方法でストリームクラスを実装する方が良いことを示しています。

ご覧のとおり、pipeメソッドは、渡されたストリームのインスタンスを返します。これにより、スレッドを結合できます。

パイプメソッドは、1つのストリームから別のストリームへのデータ転送の「速度」を制御する問題を解決するように実装されます（内部ストリームバッファーのボリュームを超える）。たとえば、Writableストリームは、Readableデータソースが渡すよりも遅い速度で書き込みます。この場合、データの次の部分を受信する準備ができたことを書き込み可能が「通知」する（内部バッファがクリアされる）まで、データ転送は「中断」されます。

バッファリング

ストリームは、内部バッファーにデータを保存します。バッファーサイズは、クラスコンストラクターで設定できるhighWaterMarkパラメーターを使用して指定できます。

highWaterMarkの物理的な意味は、別のオプション-objectModeに依存します。

 new StreamObject({objectMode: false, highWaterMark: __}); //  16384 (16kb) new StreamObject({objectMode: true, highWaterMark: __});//  16

Readableストリームでは、データはpush（data）メソッドが呼び出されるとバッファリングされ、read（）メソッドを呼び出して読み取られるまでバッファ内に残ります。読み取り可能なストリームの内部バッファーの合計サイズがhighWaterMarkで指定されたしきい値に達すると、ストリームはデータの読み取りを一時的に停止します。

書き込み可能の場合、書き込み（データ）メソッドが呼び出されるとバッファリングが発生します。このメソッドは、バッファーサイズがhighWaterMarkに達するまでtrueを返し、バッファーがいっぱいになるとfalseを返します。
pipe（）メソッドを使用する場合、その瞬間にデータの読み取りを「停止」し、「ドレイン」イベントを待機します。その後、データ転送が再開されます。

オブジェクトモード

デフォルトでは、ストリームはバッファ形式のデータを処理しますが、文字列と他のJavaScriptオブジェクトの両方を処理することもできます（たとえば、{"user"：{"name"： "Ivan"、 "last_name"： "Petrov" }}）、データの送信で別の役割を果たすnullオブジェクトを除きます（ストリームがnullを受信した場合、これは処理するデータがなく、データの読み取りまたは書き込みが完了したことを示す信号です）。初期化中にストリームの特定のモードを設定する方法は、以下の例に示されています。

ストリームの読み取り可能または一時停止状態

フロー-データは、それらを読み取るプロセスのために、継続的かつ可能な限り迅速に受信されます。
一時停止-すべてのタイプのストリームのデフォルトモード。データは明示的に要求された場合にのみ送信されます-read（）メソッドの明示的な呼び出し（read（）メソッドは暗黙的にpipe（）メソッドの「内部」と呼ばれます）

状態フロー=== true-次の場合に自動的に：

データはpipe（）メソッドを介して他のスレッドに渡されます。
および/またはイベントハンドラ「データ」を持っています;
および/またはその上でresume（）メソッドが呼び出されます。

フロー状態から一時停止状態に切り替えることができます（フロー=== false）：

データソースとそのコンシューマーの間の接続を「切断」した場合（Readable.pipe（Writable）; Readable。unpipe（Writable））、および/またはイベントハンドラー 'data'を削除します。
またはReadable.pause（）メソッドを呼び出します。

Readable flow === nullクラスの初期化時、つまり、データ読み取りメカニズムはまだ実装されておらず、データは生成されません。

読み取り可能なストリーム-データソースとしてのストリーム

読み取り可能なスレッドは、フローと一時停止の2つの状態のいずれかで動作します。一時停止状態でデータを読み取るには、明示的にread（）メソッドを呼び出す必要があります。あるストリームから別のストリームにデータを転送すると（R.pipe（W））、read（）メソッドが自動的に呼び出されます。

Readable._readableState.bufferプロパティを使用して、現在のデータバッファー全体を取得できます。

Readable.jsの例

 'use strict'; const { Readable } = require('stream'); /** *     Readable ,    _read(). *       *    (_readableState)   ,      (on('end', ()=>{})) */ class Source extends Readable { constructor(array_of_data = [], opt = {}) { super(opt); this._array_of_data = array_of_data; console.log('objectMode ', this._readableState.objectMode);//false  ,      console.log('highWaterMark ', this._readableState.highWaterMark);//16384 console.log('buffer ', this._readableState.buffer);//[] -   console.log('length ', this._readableState.length);//0 - -   console.log('flowing ', this._readableState.flowing);//null //  ,      this.on('data', (chunk)=> { //   'data' -         console.log('\n---'); console.log('Readable on data '); // chunk     console.log(`chunk = ${chunk} chunk isBuffer ${Buffer.isBuffer(chunk)} and chunk.length is ${chunk.length}`); //-     (-  ) console.log('buffer.length ', this._readableState.buffer.length); console.log(': ', chunk.toString(), ' buffer of chunk ', this._readableState.buffer, ' buffer of chunk   ', this._readableState.buffer.toString()); }) .on('error',(err)=> { console.log('Readable on error ', err); }) .on('end',()=> { console.log('Readable on end '); console.log('objectMode ', this._readableState.objectMode);//false console.log('highWaterMark ', this._readableState.highWaterMark);//16384 console.log('buffer ', this._readableState.buffer);//[] -   console.log('buffer.length ', this._readableState.buffer.length);//0 console.log('flowing ', this._readableState.flowing);//true !!!       'data' }) .on('close',()=> { console.log('Readable on close      '); }); } _read() { let data = this._array_of_data.shift() if (!data) { //,    this.push(null); } else { this.push(data); } } } /*   , ..       ,   .         this.push(data) Readable on error TypeError: Invalid non-string/buffer chunk */ let array_of_data = ['1', '2', '3', '4', '5']; let opts = {/*     */}; const R = new Source(array_of_data, opts); array_of_data = ['1', '2', '3', '4', '5']; opts = { objectMode: false, highWaterMark: 1//1      _readableState.buffer.length  === 1 }; const R2 = new Source(array_of_data, opts); array_of_data = ['1', '2', '3', '4', '5']; opts = { objectMode: false , encoding: 'utf8'//   ( NodeJS),         ,      }; const R3 = new Source(array_of_data, opts);//        .setEncoding('utf8') array_of_data = [1, 2, 3, 4, 5]; /*  ""   .  objectMode: true      -   ,    Readable.setEncoding('utf8')*/ opts = { objectMode: true , encoding: 'utf8' }; const R4 = new Source(array_of_data, opts); // objectMode: true    ,    (Number) array_of_data = [1, 2, 3, 4, 5]; opts = { objectMode: true }; const R5 = new Source(array_of_data, opts); //highWaterMark 16 -      /*     (    Writable.write(someData) === false).      Node.JS.  ,      ,    ,    */ array_of_data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]; opts = { objectMode: true }; const R6 = new Source(array_of_data, opts); R6.on('data', (chunk) => { //    1  R6.pause(); setTimeout(() => { R6.resume();//   }, 1000); });

書き込み可能なストリーム-データを書き込むためのストリーム

現在のデータバッファ全体は、writable._writableState.getBuffer（）メソッドを使用して取得できます。

Writable.jsの例

 'use strict'; const Source = require('./readable.js'); const { Writable } = require('stream'); class Writer extends Writable { constructor(opt = {}) { super(opt); console.log('objectMode ', this._writableState.objectMode);//false  ,     true console.log('highWaterMark ', this._writableState.highWaterMark);//16384 console.log('decodeStrings ', this._writableState.decodeStrings);//true  ;    Buffer ,      _write() console.log('buffer ', this._writableState.getBuffer());//[] -   this.on('drain', ()=> { console.log('\n------ writable on drain'); }) .on('error', (err)=> { console.log('\n------ writable on error', err); }) .on('finish', ()=> { console.log('\n------ writable on finish'); console.log('_writableState.getBuffer()', this._writableState.getBuffer()); }); } /** * @param chunk - || * @param encoding -   .  objectMode === true,  encoding   * @param done - callback -.     ,     ,    *   _write,  ,       chunk,      *  : done(err) -     new Error(...) * @private */ _write(chunk, encoding, done) { console.log('_writableState.getBuffer()', this._writableState.getBuffer()); console.log(typeof chunk ); //    Transform   if (typeof chunk === 'object') { console.log('chunk = ', chunk.get(), chunk.get() +' in pow '+ chunk.get() +' = '+ chunk.inPow(chunk.get())); } else { console.log(`chunk = ${chunk}; isBuffer ${Buffer.isBuffer(chunk)}; chunk.length is ${chunk.length}; encoding = ${encoding}`); } /*     . , : 1)     on('error', (err)=>{...}) 2)   ,    Readable    .         - ,  Readable.emit('error', err);     Readable.puse(),      Readable.remuse().    ,           //if (chunk > 3) return done(new Error('chunk > 3'));*/ done(); } } let array_of_data = ['1', '2', '3', '4', '5']; let r_opts = {/*    */}; const R = new Source(array_of_data, r_opts); let w_opts = {/*    */}; const W = new Writer(w_opts); R.pipe(W); array_of_data = ['1', '2', '3', '4', '5']; r_opts = {encoding: 'utf8'}; const R1 = new Source(array_of_data, r_opts); w_opts = { decodeStrings: false//  _write     'utf8',      -  (  r_opts), }; const W1 = new Writer(w_opts); R1.pipe(W1); array_of_data = [1, 2, 3, 4, 5]; r_opts = {objectMode: true}; const R2 = new Source(array_of_data, r_opts); w_opts = { objectMode: true// false,       ( r_opts),   "TypeError: Invalid non-string/buffer chunk" }; const W2 = new Writer(w_opts); R2.pipe(W2); array_of_data = [1, 2, 3, 4, 5]; r_opts = {objectMode: true}; const R3 = new Source(array_of_data, r_opts); w_opts = { objectMode: true// false,       ( r_opts),   "TypeError: Invalid non-string/buffer chunk" , highWaterMark: 1 // ;          'drain' }; const W3 = new Writer(w_opts); R3.pipe(W3); //  pipe() const R3_1 = new Source(array_of_data, r_opts); const W3_1 = new Writer(w_opts); R3_1.on('data', (chunk)=> { //R3_1._readableState.flowing === true console.log('R3_1 in flowing mode', R3_1._readableState.flowing, 'R3_1 _readableState.buffer', R3_1._readableState.buffer); toWriteOrNotToWriteThatIsTheQuestion(chunk, onDrain); }); function onDrain() { //R3_1._readableState.flowing === false,      R3_1.pause()  toWriteOrNotToWriteThatIsTheQuestion console.log('R3_1 in flowing mode', R3_1._readableState.flowing); R3_1.resume(); } /** *           Writable,       Readable (R3_1.pause()) *     ( 'drain'),       Readable ( cb R3_1.resume(); ),    Writable * @param data * @param cb */ function toWriteOrNotToWriteThatIsTheQuestion(data, cb) { // " "     write(...),    _write(...) if (!W3_1.write(data)) { R3_1.pause(); W3_1.once('drain', cb); } else { process.nextTick(cb); } }

ストリームの変換-データ変更ストリーム

変換は、デュプレックスストリームのバリエーションです。私たちは彼と最初に例を設定することにしました。

Transform.jsの例

 'use strict'; const Readable = require('./readable.js'); const Writable = require('./writable.js'); const {Transform} = require('stream'); /*  ,      , ,  JS ,    */ class Chunk { constructor(chunk) { this.set(chunk); } set(chunk) { this._chunk = chunk; } get() { return this._chunk; } inPow(pow = 2) { return Math.pow(this.get(), pow); } } class Transformer extends Transform { constructor(opt = {}) { super(opt); console.log('\n -------- Transform in constructor'); console.log('objectMode ', this._writableState.objectMode);//false  ,     true console.log('highWaterMark ', this._writableState.highWaterMark);//16384 console.log('decodeStrings ', this._writableState.decodeStrings);//true  ;    Buffer ,      _write() console.log('buffer ', this._writableState.getBuffer());//[] -   this.on('close', ()=> { console.log('\n------ Transform on close'); }) .on('drain', ()=> { console.log('\n------ Transform on drain'); }) .on('error', (err)=> { console.log('\n------ Transform on error', err); }) .on('finish', ()=> { console.log('\n------ Transform on finish'); }) .on('end', ()=> { console.log('\n------ Transform on end'); }) .on('pipe', ()=> { console.log('\n------ Transform on pipe'); }); } /** * ,      (chunk    Transform), *    -      Transform * @param chunk * @param encoding * @param done -    done(err, chunk) * @private */ _transform(chunk, encoding, done) { /*    chunk,         done(null, chunk); done(err, chunk); -       error  ,    : this.push(chunk); done(); this.push(chunk); done(err);*/ //      Chunk (.  writable.js) this.push(new Chunk(chunk)); done(); } /** *  transform     _flush.   ,      ,    'end'  Readable (  Transform,       ,    ). * @param done - done(err)     Error * @private */ _flush(done) { //TODO ... -       done(); } } let array_of_data = ['1', '2', '3', '4', '5']; let r_opts = { encoding: 'utf8' }; const R = new Readable(array_of_data, r_opts); let t_opts = { readableObjectMode: true //   Transform   , writableObjectMode: false//   Transform      , decodeStrings: false }; const T = new Transformer(t_opts); let w_opts = { objectMode: true// false,    }; const W = new Writable(w_opts); R.pipe(T).pipe(W);

デュプレックスストリーム-ストリームの書き込みと読み取り

デュプレックスは、読み取り可能、書き込み可能なストリームとしてそれ自体を実装します。さらに、それらの「作業」は互いに独立して発生します。

ストリームのトピックに興味がある場合は、Duplexストリームの実装を自分で試してみることをお勧めします。

新しいstream.Duplex（オプション）

新しいstream.Duplex（オプション）
optionsオブジェクトは、DuplexストリームのWritableおよびReadableコンストラクターに渡されます。

allowHalfOpenブール値はデフォルトでtrueです。 falseの場合、読み取りストリームが処理を完了すると、自動的に終了し、書き込みストリームが終了します。
readableObjectModeブール値はデフォルトでfalseです。読み取り可能なストリームのobjectModeモード。プロパティobjectMode = trueの場合、プロパティの値は無視されます。
writableObjectMode booleanはデフォルトでfalseです。書き込み可能なストリームのobjectModeモード。プロパティobjectMode = trueの場合、プロパティの値は無視されます。

エラー処理

一部のリンクで「エラー」イベントがトリガーされ、チェーン内の「前の」ストリームに通知する必要がある場合、「エラー」イベントを呼び出す必要もあります：StreamClass.emit（ 'error'、err）状況を処理します。または、この問題を解決できるポンプモジュール（https://github.com/mafintosh/pump）を使用します。

まとめると

スレッドを使用すると、ほとんどすべての問題を解決できます。

ファイルまたはデータベースへのデータの保存/読み取り;
ファイルのアーカイブ。
データ暗号化;
ネットワークを介したデータ伝送（リアルタイムメッセージングモジュール、ビデオ、オーディオファイルブロードキャストの実装）

彼らが言うように-すべての味のために。

NodeJSのストリーム-2回入る川