Node streamning
はじめに
業務で容量が大きいファイルの I/O 操作を Node で行う機会があった。
以前は、ファイル全体をメモリに展開して、処理を行っていたが大量のデータの場合
メモリリークを起こしてしまったり処理が重くなるため、ストリームを学んだ。
ファイル操作以外にもできることはたくさんあるが、今回は一部学んだ内容について
頭の中を整理するために記事に書く。
環境
Node:22.9.0
Typescript:5.1.3
Node Stream とは
大量のデータを効率的に処理するための強力な仕組み
データをチャンクに分けて順次処理する。これにより、大量のデータを
一度にメモリに読み込む必要がなくなり、メモリ使用量の削減や処理の高速化が可能になる。
ストリームには主にReadable,Writable,Duplex,Transformが存在している。
今回は、ReadableとWritableのみ使用したためDuplexとTransformについては書かない。
記事を読む限り、Duplexは、Readable+WritableでTransformは、読み取りや書き込みの時に
データを変換できたりするらしい。
Readable
Readable ストリームは、外部のデータソース（ファイル、HTTP リクエスト、データベースなど）からデータを読み込むストリーム。
全てのリソースを読みこむのではなくチャンクと呼ばれるデータの塊ごとに読み込んでいく。
stream.Readable クラスを使用する。
Readable ストリームには、flowing モード と non-flowing モード という 2 つの動作モードがある。
flowing モード
リソースからストリームに流れてきたデータを自動的に読み取る。
読み取るとすぐにdataイベントが発火し、データが消費されていく。
dataイベントのリスナーに処理を渡すこと流れてきたデータに対して処理を行うことができる。
non-flowing モード
データは自動的には流れず、明示的に stream.read() メソッドを呼び出してデータを読み取る。
手動で読み取りを制御するには、createReadStream 関数を使用する際に、highWaterMark オプションを指定する。
HighWaterMark
ストリーム内の閾値のこと。
データを内部バッファに取り込む時に閾値を超える量が流れてくると一時ストップする。
閾値を決めていないとバッファにデータが溜まりメモリリークを起こしてしまう。
このような、ストリームの流用制御をすることをバックプレッシャーと呼ぶ。
Whitable
Whitable ストリームは、データの書き込みに使用する。
こちらも同様に内部バッファを使用しデータをため出力先にデータを流していく。
HighWaterMarkも設定することが可能。
ストリームイベント
以下は、ストリームのイベント。
公式を見たら結構な量あるため自身が使用したイベントだけ書く。
基本は、ストリームのイベントにリスナーを設定していく。
Readable ストリームのイベント
・data イベント
ストリームの内部バッファにデータが入った時はしる。
・end イベント
ストリームから取り出せるデータが無くなった時はしる。
・close イベント
読み込んでいるリソースを閉じた時はしる。
・error イベント
ストリーム中にエラーが発生した時にはしる。
Whitable ストリームのイベント
・write イベント
リソースに出力するイベント。
このイベントは、リスナーを設定するのではなく、Whitable ストリームの
インスタンスから呼び出す。
・close イベント
リソースを閉じた時はしる。
・error イベント
ストリーム中にエラーが発生した時にはしる。
今後
実装時に使わなかったが、Pipe という概念がある。
簡単に言えば、Readable Stream と Writable Stream の橋渡しをしてくれるもの。
この中でデータを少し加工したい時は Transform などを使用したりする。
この辺は、必要になってきたらおいおい調べていこうと思う。
引用
Discussion