Node.js Streams: پردازش داده‌های بزرگ

فایل بزرگ لازم نیست یک‌باره وارد حافظه شود

اگر برای تغییر هر سطر یک فایل، ابتدا تمام فایل را بخوانیم و سپس split کنیم، چند نسخه از داده ممکن است هم‌زمان در حافظه باشند. Stream امکان می‌دهد بخش‌های داده را در طول زمان بخوانیم، تبدیل کنیم و بنویسیم. این الگو برای فایل بزرگ مفید است، اما تضمین مصرف حافظهٔ ثابت یا عدد خاصی مثل ده مگابایت نمی‌دهد. Bufferها، رشته‌ها، خروجی transform و کدی که داده را جمع می‌کند، همگی روی مصرف حافظه اثر دارند. روش پردازش از نام ابزار مهم‌تر است.

در یک pipeline مناسب، بخش خواندن با سرعت بخش نوشتن هماهنگ می‌شود. اگر مقصد کندتر باشد، نباید کل دادهٔ باقیمانده بی‌وقفه در آرایه یا صف برنامه جمع شود. همین هماهنگی با نام backpressure شناخته می‌شود. Stream آن را در API فراهم می‌کند، اما کد مصرف‌کننده هم باید قرارداد را رعایت کند. مثال این نوشته متن UTF-8 را سطر به سطر تبدیل می‌کند و محدودیت طول سطر دارد؛ parser عمومی CSV یا پردازش همهٔ قالب‌ها نیست.

Readable، Writable و Transform

Readable داده فراهم می‌کند و Writable داده می‌پذیرد. Duplex هر دو سمت را دارد و Transform نوعی stream دوطرفه برای تبدیل ورودی به خروجی است. فایل ورودی نمونه Readable، فایل خروجی Writable و تغییر متن Transform است. این نام‌ها رفتار عمومی را بیان می‌کنند؛ هر stream باید خطا، پایان و منابع خودش را هم درست مدیریت کند. مرجع Stream در Node.js APIهای این اجزا و pipeline را توضیح می‌دهد.

Object mode و byte stream قرارداد یکسانی ندارند. در حالت معمول داده Buffer یا رشته است؛ در object mode هر item می‌تواند یک object باشد. اندازهٔ صف در این دو حالت با واحدهای متفاوتی معنا پیدا می‌کند. تبدیل هزاران record به object و نگه داشتن reference آن‌ها در جای دیگر، می‌تواند حافظه را زیاد کند؛ حتی وقتی ورودی stream است. پیش از انتخاب حالت، معلوم کنید stage بعدی bytes می‌خواهد، متن می‌خواهد یا record کامل.

Backpressure را از مقصد شروع کنید

در نوشتن دستی به Writable، false از write نشانه‌ای است که باید تا آماده شدن مقصد منتظر بمانید. نادیده گرفتن آن می‌تواند صف نوشتن را بزرگ کند. Pipe و pipeline برای اتصال‌های استاندارد این هماهنگی را مدیریت می‌کنند. راهنمای backpressure در Node.js اثر رعایت و نادیده گرفتن قرارداد را توضیح می‌دهد. عددهای آزمایش آن راهنما را نباید به عنوان مصرف تضمین‌شدهٔ برنامهٔ دیگری نقل کرد.

highWaterMark آستانهٔ buffer است و سقف قطعی کل حافظهٔ process نیست. یک chunk، رشتهٔ decodeشده و خروجی تبدیل می‌توانند هم‌زمان وجود داشته باشند. برخی تبدیل‌ها هم از ورودی، خروجی بزرگ‌تری می‌سازند. بنابراین کاهش مقدار به عدد کوچک، لزوماً بهترین تنظیم نیست و ممکن است هزینهٔ پردازش و فراخوانی را زیاد کند. ابتدا شکل داده و سرعت مقصد را بشناسید. برای نمونهٔ آموزشی، مقدار خواندن صریح انتخاب شده تا واحد آن روشن بماند، نه اینکه بهترین مقدار همهٔ پروژه‌ها معرفی شود.

Chunk مرز سطر یا کاراکتر نیست

خواندن فایل در chunkهای جدا به معنی دریافت یک سطر در هر chunk نیست. یک سطر ممکن است میان دو chunk شکسته شود و چند سطر هم در یک chunk باشند. کاراکتر UTF-8 چندبایتی نیز ممکن است میان bufferها تقسیم شود. استفاده از encoding مناسب در ReadStream به decode پیوستهٔ متن کمک می‌کند؛ تبدیل مستقل هر Buffer به رشته بدون توجه به مرز می‌تواند مشکل بسازد. مستندات StringDecoder دلیل نگه داشتن بخش ناقص کاراکتر را بیان می‌کند.

برای سطر، بخشی که هنوز newline ندارد در carry می‌ماند. اگر فایل یک سطر بسیار بزرگ داشته باشد، همین carry می‌تواند بخش عمدهٔ فایل شود. بنابراین محدود کردن طول record، بخشی از کنترل حافظه است. در نمونه، طول سطر بر اساس واحد UTF-16 رشتهٔ JavaScript سنجیده می‌شود؛ این عدد دقیقاً تعداد byte یا تعداد کاراکتر دیداری نیست. CRLF هم هنگام پایان سطر به LF تبدیل می‌شود. قالبی که newline داخل مقدار quoted دارد، مانند بعضی فایل‌های CSV، parser جدا می‌خواهد.

نمونهٔ کامل با ورودی و خروجی مستقل

فایل اول را با نام lines.mjs و فایل متن را با نام input.txt در یک پوشه ذخیره کنید. Node 24 یا نسخهٔ سازگار جدیدتر لازم است و package خارجی نداریم. فرمان node lines.mjs input.txt output.txt پردازش را شروع می‌کند. مقصد output.txt باید تازه باشد؛ flags برابر wx اجازهٔ بازنویسی فایل موجود را نمی‌دهد. این انتخاب جلوی جایگزین شدن اتفاقی ورودی یا خروجی قبلی را می‌گیرد. نام مسیرها را متناسب با فایل‌های خود تغییر دهید و برای مسیر دارای فاصله از نقل‌قول مناسب shell استفاده کنید.

import { createReadStream, createWriteStream } from 'node:fs';
import { Transform } from 'node:stream';
import { pipeline } from 'node:stream/promises';

class UppercaseLines extends Transform {
  constructor(maxLine = 8192) {
    super({ decodeStrings: false });
    this.carry = '';
    this.maxLine = maxLine;
    this.lines = 0;
  }
  emitLine(line) {
    const text = line.endsWith('\r') ? line.slice(0, -1) : line;
    if (text.length > this.maxLine) throw new Error('Line exceeds allowed UTF-16 length');
    this.lines += 1;
    this.push(text.toUpperCase() + '\n');
  }
  _transform(chunk, encoding, callback) {
    try {
      const parts = (this.carry + chunk).split('\n');
      this.carry = parts.pop();
      for (const line of parts) this.emitLine(line);
      if (this.carry.length > this.maxLine + 1) throw new Error('Unfinished line exceeds limit');
      callback();
    } catch (error) { callback(error); }
  }
  _flush(callback) {
    try {
      if (this.carry.length) this.emitLine(this.carry);
      callback();
    } catch (error) { callback(error); }
  }
}

const [source, destination] = process.argv.slice(2);
if (!source || !destination) {
  console.error('Usage: node lines.mjs input.txt output.txt');
  process.exitCode = 1;
} else {
  const controller = new AbortController();
  const cancel = () => controller.abort();
  process.once('SIGINT', cancel);
  const transform = new UppercaseLines();
  try {
    await pipeline(
      createReadStream(source, { encoding: 'utf8', highWaterMark: 64 * 1024 }),
      transform,
      createWriteStream(destination, { flags: 'wx' }),
      { signal: controller.signal },
    );
    console.log(`Processed lines: ${transform.lines}`);
  } catch (error) {
    console.error(`Processing stopped: ${error.message}`);
    console.error('An incomplete output may remain; choose a new destination for the next run.');
    process.exitCode = 1;
  } finally {
    process.removeListener('SIGINT', cancel);
  }
}

hello streams
سلام به پردازش مرحله‌ای
one line at a time

Transform حروف دارای حالت بزرگ و کوچک را به uppercase تبدیل می‌کند و متن فارسی در این تبدیل معمولاً همان شکل می‌ماند. هدف نمونه، تغییر معنایی زبان نیست؛ مسیر خواندن، مرز سطر و خروجی مرحله‌ای است. هر سطر خروجی LF دارد و اگر آخرین سطر ورودی newline نداشته باشد، خروجی برای آن newline اضافه می‌کند. سطرهای خالی میانی حفظ می‌شوند. ورودی آموزشی کوتاه است و آمار مصرف حافظه یا زمان اجرای فایل بزرگ از آن نتیجه نمی‌گیریم. این مقاله روش اجرا را می‌دهد و اجرای آزمایش گزارش نشده است.

متد emitLine طول سطر نهایی را بررسی می‌کند و سپس خروجی را push می‌کند. پردازش یک chunk هم‌زمان انجام می‌شود؛ خروجی همان chunk می‌تواند موقتاً صف ایجاد کند، اما chunkهای بعدی در اتصال stream با backpressure هماهنگ می‌شوند. این نکته مهم است: نمونه ادعای سقف دقیق حافظه ندارد. محدودیت carry از رشد نامحدود یک سطر ناتمام جلوگیری می‌کند. برای تبدیل‌هایی با گسترش بسیار زیاد خروجی، باید طراحی تولید تدریجی خروجی دقیق‌تر باشد.

Pipeline خطا را در یک نقطه جمع می‌کند

خواندن ورودی ممکن است به دلیل نبود فایل شکست بخورد؛ نوشتن ممکن است به دلیل مجوز یا پر شدن دیسک متوقف شود؛ transform هم می‌تواند دادهٔ نامعتبر پیدا کند. await روی pipeline نتیجهٔ کل زنجیره را قابل مشاهده می‌کند. catch پیام شکست و کد خروج ناموفق را ثبت می‌کند. خطا را صرفاً log نمی‌کنیم و بعد نتیجهٔ موفق نمی‌دهیم. این قرارداد برای استفاده در script یا CI اهمیت دارد، چون مرحلهٔ بعد باید بتواند شکست را تشخیص دهد.

خروجی ناموفق ممکن است ناقص باقی بماند. نمونه آن را خودکار حذف نمی‌کند و صریحاً می‌گوید مقصد جدید انتخاب کنید. در محصول واقعی می‌توان نوشتن در فایل موقت و انتقال پس از موفقیت را طراحی کرد؛ اما مسیر موقت، مالکیت فایل و محدودیت rename در محیط مقصد باید روشن باشند. همچنین وجود output قبلی به معنی کامل بودن آن نیست. مصرف‌کننده باید معیار موفقیت را از پایان معتبر پردازش یا قرارداد انتشار بگیرد، نه فقط از نام فایل.

لغو با پایان عادی متفاوت است

در نمونه، SIGINT به AbortController وصل شده و signal به pipeline می‌رسد. لغو موجب توقف مسیر می‌شود و موفقیت عادی گزارش نمی‌شود. Listener در finally آزاد می‌شود تا عمر آن مشخص باشد. این روش برای اجرای command-line ساده است؛ در server باید signal به عمر درخواست یا کار مرتبط باشد. لغو عملیات، تضمین بازگشت اثرهای قبلی نیست. داده‌ای که پیش از لغو نوشته شده، ممکن است در فایل ناقص وجود داشته باشد. نوشتهٔ Promise و لغو صریح این تفاوت را توضیح می‌دهد.

کار CPU همچنان می‌تواند حلقهٔ رویداد را مشغول کند

Stream داده را مرحله‌ای می‌رساند، ولی یک transform سنگین هم‌زمان می‌تواند event loop را برای مدتی درگیر کند. فشرده‌سازی، parse پیچیده یا محاسبهٔ طولانی نیازمند شناخت هزینهٔ پردازش است. اضافه کردن async به نام تابع، کار CPU را خودکار به thread دیگری منتقل نمی‌کند. برای کارهای متفاوت می‌توان worker یا ابزار مناسب را بررسی کرد. نمونهٔ uppercase کار محدود دارد؛ دربارهٔ توان پاسخ‌دهی server در بار واقعی ادعایی ندارد.

خروجی را به مصرف‌کنندهٔ واقعی وصل کنید

اگر پس از stream تمام نتیجه را در یک آرایه جمع کنید، مزیت پردازش مرحله‌ای ممکن است از بین برود. اگر هر سطر را به API بفرستید و درخواست‌های نامحدود بسازید، صف شبکه مسئلهٔ تازه‌ای می‌شود. تعداد عملیات هم‌زمان و سیاست retry باید مشخص باشد. برای build script، خروجی و کد خروج را در مسیر CI/CD به مرحلهٔ بعد وصل کنید. در محیط container نیز حجم حافظه و دسترسی دیسک بخشی از تنظیم واقعی است؛ مقالهٔ Docker فرانت‌اند تفاوت محیط و ابزار را توضیح می‌دهد.

پیش از انتخاب Stream، شکل record، بیشترین اندازهٔ مجاز، مقصد و رفتار در خطا را ثبت کنید. اگر فایل کوچک و محدود است، خواندن کامل می‌تواند انتخاب ساده و مناسبی باشد. وقتی حجم داده یا سرعت مقصد نیازمند پردازش مرحله‌ای است، Stream همراه با backpressure و محدودیت record مسیر قابل نگهداری فراهم می‌کند. حذف ادعای حافظهٔ ثابت کمک می‌کند این تصمیم بر اساس دادهٔ واقعی و قرارداد برنامه گرفته شود.