ストリーム、pipeline、バックプレッシャーのおかげで、数ギガバイトのファイルを一定のメモリで処理しましょう。
このレッスンを Kodokon で開くreadFileはファイル全体をメモリに読み込みます。4 GBのログでは、あなたのプロセスは破裂してしまいます。ストリームはデータをチャンク単位で処理し、highWaterMark（ファイルストリームではデフォルトで64 KiB）に制限された一定のメモリ使用量で済みます。四つの型があります。Readable、Writable、Duplex、Transformです。これらが解決する中心的な問題はバックプレッシャーと呼ばれます。生産者が消費者より速いとき、どうしますか？
import { createWriteStream } from "node:fs";
const out = createWriteStream("big.txt");
let i = 0;
function writeChunks() {
let ok = true;
while (i < 1e6 && ok) {
ok = out.write(`line ${i}\n`);
i += 1;
}
if (i < 1e6) out.once("drain", writeChunks);
else out.end();
}
writeChunks();write()がfalseを返すとき、内部バッファはhighWaterMarkを超えています。データは失われませんが、書き込みを続けるとすべてがメモリに溜まっていきます。これはまさにストリームが避けようとしていたことです。規約はこうです。書き込みをやめてdrainイベントを待つのです。これはpipe()があなたの代わりにやってくれることであり、pipeline()はそれをさらにうまくやります。バックプレッシャーに加えて、エラーを伝播させ、すべてのストリームを両方向にきれいに破棄します。
import { createReadStream, createWriteStream }
from "node:fs";
import { Transform } from "node:stream";
import { pipeline } from "node:stream/promises";
const upper = new Transform({
transform(chunk, encoding, callback) {
callback(null, chunk.toString().toUpperCase());
},
});
await pipeline(
createReadStream("big.txt"),
upper,
createWriteStream("big-upper.txt"),
);
console.log("done");Readableストリームは非同期イテラブルです。for awaitはバックプレッシャーを尊重しながらチャンクを消費します。気をつけてください。チャンクはバイトの任意の切れ端で、それが行の境目で終わる保証はどこにもありません。行ごとの処理には、node:readlineが断片の再結合を引き受けてくれます。crlfDelay: Infinityのおかげで、二つのチャンクにまたがって分割された\r\nも扱えます。
import { createReadStream } from "node:fs";
import { createInterface } from "node:readline";
const rl = createInterface({
input: createReadStream("big.txt"),
crlfDelay: Infinity,
});
let count = 0;
for await (const line of rl) {
if (line.includes("42")) count += 1;
}
console.log(count, "matching lines");