التدفّقات (Streams) والضغط العكسي: التعامل مع الملفات الكبيرة

عالِج ملفات بحجم عدة غيغابايت بذاكرة ثابتة بفضل التدفّقات، وpipeline، والضغط العكسي.

9 دقيقة · 3 أسئلة

تُحمّل readFile ملفًا بأكمله في الذاكرة: فمع سجلّ حجمه 4 غيغابايت، تنهار عمليتك. أما التدفّقات فتعالج البيانات على شكل أجزاء (chunks) ببصمة ذاكرة ثابتة، محدودة بـ highWaterMark (64 كيبيبايت افتراضيًا لتدفّق ملف). وتوجد أربعة أنواع: Readable، وWritable، وDuplex، وTransform. والمشكلة الجوهرية التي تحلّها تُسمى الضغط العكسي (backpressure): ماذا تفعل عندما يكون المُنتِج أسرع من المُستهلِك؟

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 الأجزاء مع احترام الضغط العكسي. لكن انتبه، فالجزء (chunk) شريحة اعتباطية من البايتات: لا شيء يضمن أنه ينتهي عند حدّ سطر. وللمعالجة سطرًا سطرًا، تتولّى node:readline إعادة تجميع القطع، بما في ذلك \r\n مقسومًا على جزأين بفضل crlfDelay: Infinity.

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؟
   • عدد الكائنات المُنتظِرة في الطابور
   • عدد من البايتات، كما في الوضع الثنائي
   • الحجم الأقصى لكائن مُسلسَل