स्ट्रीम और बैकप्रेशर: बड़ी फ़ाइलों को संभालना

स्ट्रीम, pipeline और बैकप्रेशर की बदौलत कई गीगाबाइट की फ़ाइलों को स्थिर मेमोरी के साथ प्रोसेस करें।

9 मिनट · 3 प्रश्न

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 स्ट्रीम एसिंक्रोनस iterables हैं: 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");

ज्ञान जांच

सुनिश्चित करें कि आपको इस पाठ के मुख्य बिंदु याद हैं।

1. जब writable.write(chunk) false लौटाता है तो इसका क्या अर्थ है?
   • चंक खो गया और उसे दोबारा लिखना होगा
   • आंतरिक बफ़र highWaterMark से आगे बढ़ गया है: drain इवेंट तक लिखना बंद करें
   • स्ट्रीम किसी त्रुटि से नष्ट हो गई
2. pipeline() को pipe() पर कौन सा निर्णायक लाभ है?
   • pipeline तेज़ है क्योंकि यह मल्टीथ्रेडेड है
   • pipe बैकप्रेशर संभालता है, pipeline नहीं
   • pipeline त्रुटियों को आगे प्रसारित करता है और इसमें शामिल सभी स्ट्रीम को नष्ट कर देता है
   • pipeline स्वतः एन्कोडिंग रूपांतरित करता है
3. objectMode में, highWaterMark क्या दर्शाता है?
   • कतारबद्ध ऑब्जेक्ट की एक संख्या
   • बाइट्स की एक संख्या, जैसा कि बाइनरी मोड में होता है
   • किसी सीरियलाइज़्ड ऑब्जेक्ट का अधिकतम आकार