Backpressure في Node.js: عالج 2GB بدون تفجير الذاكرة
مستوى القارئ: متوسط
هتكسب من المقال ده إنك تعرف تعالج ملفات كبيرة أو streams بطيئة في Node.js بدون ما الذاكرة تطلع لحد OOM. ركز: المشكلة غالبًا مش إن السيرفر ضعيف، المشكلة إن الكود بيسمح للبيانات تتراكم أسرع من استهلاكها.
المشكلة باختصار
عندك سكربت بياخد ملف log حجمه 2GB، يفلتر السطور المهمة، وبعدها يكتب النتيجة في ملف تاني أو يبعتها لخدمة HTTP. الطريقة السهلة هي fs.readFile، لكنها بتحمّل الملف كله في الذاكرة. الطريقة اللي شكلها أذكى هي إنك تستخدم stream، لكن لو كتبت chunks بسرعة في وجهة أبطأ بدون احترام drain، أنت عملت queue كبير في الذاكرة بدل ما حلّيت المشكلة.
اللي بيحصل فعلاً: الـ Readable بيطلع بيانات بسرعة، والـ Writable مش قادر يكتب بنفس السرعة. لو تجاهلت الإشارة دي، الذاكرة تزيد بهدوء. في اختبار عملي على ملف 2GB، readFile ممكن يوصل لاستهلاك قريب من 2100MB، وكتابة stream يدوية غلط ممكن تقف عند 820MB، بينما pipeline مع backpressure يحافظ على حدود حوالي 74MB. الأرقام تقديرية لاختبار واحد، لكنها توضح الاتجاه.
مثال بسيط قبل التعريف العلمي
اعتبر عندك موظف بيجهز طلبات بسرعة، وموظف تاني بيلفّها في كراتين ببطء. لو الأول فضل يرمي الطلبات على الترابيزة، المكان هيمتلئ. الحل مش إننا نطلب من الموظف التاني يسرع دائمًا. الحل إن الأول يقف لحظة لما الترابيزة توصل للحد المسموح، ويرجع يشتغل لما المساحة تفضى.
ده بالظبط backpressure. في Node.js، الـ buffer هو الترابيزة، وhighWaterMark هو الحد اللي بعده Node يقول: استنى. لما writable.write(chunk) ترجع false، دي إشارة إن الوجهة مش جاهزة للمزيد. لما يحصل حدث drain، تقدر تكمل.
أفضل طريقة: استخدم pipeline لما تقدر
لو عندك سلسلة واضحة: قراءة ملف، Transform، كتابة ملف أو ضغطه، استخدم stream/promises وpipeline. المكسب: backpressure وإدارة أخطاء وتنظيف موارد في مسار واحد. الـ trade-off هنا إن الكود أقل مرونة من التحكم اليدوي الكامل، لكنه أفضل لمعظم الحالات الإنتاجية.
import { createReadStream, createWriteStream } from 'node:fs';
import { Transform } from 'node:stream';
import { pipeline } from 'node:stream/promises';
const onlyErrors = new Transform({
transform(chunk, encoding, callback) {
const lines = chunk.toString('utf8').split('\n');
const filtered = lines.filter(line => line.includes('ERROR')).join('\n');
callback(null, filtered ? filtered + '\n' : '');
}
});
await pipeline(
createReadStream('app.log', { highWaterMark: 64 * 1024 }),
onlyErrors,
createWriteStream('errors.log')
);
console.log('done');