Архітектура середовища Node.js спроектована для обробки мережевого введення-виведення з мінімальним споживанням апаратних ресурсів. У монолітній моделі буферизації (наприклад, при зчитуванні файлу через fs.readFile) весь файл завантажується в оперативну пам'ять процесу. Якщо сервер одночасно обслуговує 100 клієнтських запитів на вивантаження файлів по 500 МБ, застосунку знадобиться 50 ГБ RAM, що неминуче спричинить падіння процесу з помилкою JavaScript heap out of memory або аварійну зупинку через OOM Killer ядра операційної системи.
Потоки даних (Streams) реалізують фундаментальний принцип потокової обробки: розбиття неперервного потоку байтів на дискретні фрагменти (чанки — chunks) та їх послідовну обробку по мірі надходження. Споживання пам'яті обмежується лише розміром активних чанків (порядку кількох десятків кілобайтів) незалежно від того, передається файл розміром 100 МБ чи 100 ГБ, або скільки часу триває мережеве з'єднання.
В основі всіх потоків лежить модуль node:stream, що базується на патерні EventEmitter та зв'язаному списку буферів BufferList. У середовищі виконання виділяються чотири базові класи:
Readable: джерело даних, з якого здійснюється читання байтів (файловий потік fs.createReadStream, тіло вхідного HTTP-запиту http.IncomingMessage, мережевий сокет).Writable: приймач даних, у який записуються байти (файловий потік fs.createWriteStream, вихідна HTTP-відповідь http.ServerResponse, стандартний вивід process.stdout).Duplex: двонаправлений потік із незалежними каналами читання та запису (наприклад, двосторонній TCP-сокет net.Socket).Transform: спеціалізований підклас Duplex, у якому дані на виході обчислюються з вхідних даних (наприклад, стиснення zlib.createGzip(), потокове шифрування crypto.createCipheriv() або синтаксичні аналізатори).Кожен потік інкапсулює внутрішній стан: _readableState для потоків читання та _writableState для потоків запису. Ключовим параметром конфігурації є highWaterMark:
objectMode: true) ліміт дорівнює 16 дискретним JavaScript-об'єктам.highWaterMark — це не жорсткий ліміт пам'яті, а сигнальний поріг. Якщо приймач повільніший за джерело, байти накопичуватимуться в пам'яті понад highWaterMark, якщо не активовано протокол зворотного тиску.
Явище Backpressure виникає в будь-якій асинхронній системі, коли темп генерації даних джерелом (producer) перевищує швидкість їх обробки споживачем (consumer) або пропускну здатність каналу запису:
Темп генерації джерела (Producer) >> Швидкість споживання (Consumer)
(наприклад: швидкий диск 3 ГБ/с) (наприклад: мобільний 3G клієнт 50 КБ/с)Без регулювання різниці швидкостей черга запису розростеться в оперативній пам'яті. У Node.js Backpressure працює за двофазним протоколом:
writable.write(chunk) повертає булеве значення. Якщо розмір черги приймача перевищує highWaterMark, write() повертає false. Це директива джерелу: негайно призупинити зчитування (readable.pause()).'drain'. Джерело отримує сигнал і відновлює генерацію (readable.resume()).Метод readable.pipe(writable) тривалий час був основним способом зв'язування потоків, але в промислових системах вважається антипатерном з таких причин:
pipe() не поширює події помилок ланцюжком. Якщо проміжний потік або приймач зазнає збою, інші потоки залишаються відкритими. Файлові дескриптори блокуються, вичерпуючи системні ліміти EMFILE: too many open files.pipe() не підтримує AbortController і не може бути безпечно перерваний при закритті з'єднання клієнтом.'error', 'close' та 'finish' для кожної ланки.Сучасним стандартом є функція stream.promises.pipeline, яка гарантує атомарність: при помилці в будь-якому потоці або отриманні AbortSignal всі потоки в ланцюжку негайно знищуються (destroy(err)), закриваючи дескриптори та звільняючи ресурси для збирача сміття.
Роботу механізму Backpressure можна зобразити як шлюзовий механізм з автоматичним клапаном тиску:
[ Readable Stream ] (Джерело: Диск 500 МБ/с)
|
| 1. chunk (64 КБ)
v
[ Writable Stream ] (Внутрішній буфер: highWaterMark = 64 КБ)
+-------------------------------------------------------+
| Стан черги: [ XXXXXXXX XXXXXXXX XXXXXXXX XXXXXXXX ] | -> Буфер повний!
+-------------------------------------------------------+
|
+---> writable.write(chunk) === false
|
v
[ readable.pause() ] <--- Джерело ПРИЗУПИНЯЄ читання
|
(Ядро ОС повільно передає TCP-пакети клієнту...)
|
+-------------------------------------------------------+
| Буфер звільнено: [ ] |
+-------------------------------------------------------+
|
v
[ emit('drain') ]
|
+---> [ readable.resume() ] <--- Джерело ВІДНОВЛЮЄ передачу!zlib.createBrotliCompress()), що динамічно підлаштовується під швидкість з'єднання клієнта.Промисловий потік Transform приймає бінарний потік CSV, обробляє можливі розриви рядків на межах чанків (chunk boundary issue) та емітить валідні JavaScript-об'єкти:
import { Transform, TransformCallback } from 'node:stream';
export interface CsvTransformOptions {
delimiter?: string;
hasHeader?: boolean;
}
export class CsvToJsonTransform extends Transform {
private delimiter: string;
private hasHeader: boolean;
private headers: string[] = [];
private leftover: string = '';
private isFirstChunk: boolean = true;
constructor(options: CsvTransformOptions = {}) {
super({
readableObjectMode: true, // На виході видаємо JS-об'єкти
writableObjectMode: false, // На вході приймаємо бінарні буфери
highWaterMark: 64 * 1024, // Пороговий буфер 64 КБ
});
this.delimiter = options.delimiter ?? ',';
this.hasHeader = options.hasHeader ?? true;
}
/**
* Викликається для кожного вхідного чанка даних.
*/
public _transform(chunk: Buffer, encoding: BufferEncoding, callback: TransformCallback): void {
try {
// Об'єднуємо залишок попереднього чанка з поточним
const content = this.leftover + chunk.toString('utf-8');
const lines = content.split(/\r?\n/);
// Останній елемент може бути неповним рядком через межу чанка
this.leftover = lines.pop() ?? '';
for (const line of lines) {
const trimmed = line.trim();
if (!trimmed) continue;
const columns = trimmed.split(this.delimiter).map((col) => col.trim());
if (this.isFirstChunk && this.hasHeader) {
this.headers = columns;
this.isFirstChunk = false;
continue;
}
const record: Record<string, string> = {};
for (let i = 0; i < columns.length; i++) {
const key = this.headers[i] ?? `col_${i}`;
record[key] = columns[i];
}
// Передаємо розібраний запис далі по ланцюжку
this.push(record);
}
callback(); // Готовність до наступного чанка
} catch (err) {
callback(err instanceof Error ? err : new Error(String(err)));
}
}
/**
* Викликається один раз після вичерпання джерела даних.
*/
public _flush(callback: TransformCallback): void {
try {
if (this.leftover.trim()) {
const columns = this.leftover.trim().split(this.delimiter).map((col) => col.trim());
const record: Record<string, string> = {};
for (let i = 0; i < columns.length; i++) {
const key = this.headers[i] ?? `col_${i}`;
record[key] = columns[i];
}
this.push(record);
}
callback();
} catch (err) {
callback(err instanceof Error ? err : new Error(String(err)));
}
}
}У сучасному Node.js читання Readable потоків декларативно виконується через цикл for await (const chunk of stream). Середовище синхронізує зворотний тиск із кожною ітерацією циклу:
import { createReadStream, createWriteStream } from 'node:fs';
import { pipeline } from 'node:stream/promises';
import { Transform } from 'node:stream';
async function processBigDataset(sourcePath: string, targetPath: string, signal: AbortSignal): Promise<void> {
const readStream = createReadStream(sourcePath, { highWaterMark: 64 * 1024 });
const writeStream = createWriteStream(targetPath);
// Кастомний Transform для маскування конфіденційних даних
const anonymizer = new Transform({
transform(chunk: Buffer, encoding, callback) {
const sanitized = chunk.toString('utf-8').replace(/\b\d{4}-\d{4}-\d{4}-\d{4}\b/g, 'XXXX-XXXX-XXXX-XXXX');
callback(null, Buffer.from(sanitized));
}
});
try {
await pipeline(
readStream,
anonymizer,
writeStream,
{ signal } // Передача AbortSignal для безпечного переривання
);
console.log('Конвеєр успішно завершено без витоку дескрипторів');
} catch (error) {
if ((error as { name?: string }).name === 'AbortError') {
console.warn('Обробку файлу зупинено за сигналом відміни');
} else {
console.error('Критичний збій у ланцюжку потоків:', error);
throw error;
}
}
}Потік читання завантажує блок 64 КБ із диска у внутрішню чергу пам'яті.
Transform-потік розбирає рядки, санітизує поля та формує серіалізовані NDJSON-об'єкти.
writable.write() повертає false при досягненні highWaterMark. Джерело переходить у pause().
Мережевий стек операційної системи передає TCP-пакети клієнту, звільняючи чергу.
Writable емітить подію drain. Джерело викликає resume() та читає наступний фрагмент.
Промисловий потоковий процесор журналів подій із контролем зворотного тиску, санітизацією даних, стисненням Gzip та коректним завершенням:
import { Readable, Writable, Transform, TransformCallback } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { createGzip } from 'node:zlib';
export interface LogEntry {
timestamp: string;
level: 'info' | 'warn' | 'error' | 'fatal';
service: string;
message: string;
context?: Record<string, unknown>;
}
export interface ProcessorMetrics {
totalProcessed: number;
filteredOut: number;
bytesWritten: number;
backpressurePauses: number;
}
export class HighThroughputLogProcessor {
private metrics: ProcessorMetrics = {
totalProcessed: 0,
filteredOut: 0,
bytesWritten: 0,
backpressurePauses: 0,
};
/**
* Створює Transform-потік для валідації та санітизації записів журналу.
*/
private createSanitizerTransform(): Transform {
const self = this;
return new Transform({
objectMode: true,
highWaterMark: 256, // 256 об'єктів у черзі
transform(entry: LogEntry, encoding: BufferEncoding, callback: TransformCallback) {
self.metrics.totalProcessed++;
// Фільтрація некритичних записів health-check
if (entry.level === 'info' && entry.message.includes('HEALTHCHECK')) {
self.metrics.filteredOut++;
return callback(); // Пропускаємо запис без виклику push
}
// Санітизація конфіденційних полів (PII)
if (entry.context) {
if ('password' in entry.context) entry.context.password = '[REDACTED]';
if ('token' in entry.context) entry.context.token = '[REDACTED]';
if ('authorization' in entry.context) entry.context.authorization = '[REDACTED]';
}
// Серіалізація в формат NDJSON
const serialized = JSON.stringify(entry) + '\n';
callback(null, Buffer.from(serialized, 'utf-8'));
},
});
}
/**
* Створює інструмент моніторингу Backpressure у потоці запису.
*/
private createMetricsMonitor(targetWritable: Writable): Writable {
const self = this;
const monitored = new Writable({
highWaterMark: 64 * 1024,
write(chunk: Buffer, encoding: BufferEncoding, callback) {
self.metrics.bytesWritten += chunk.byteLength;
const canContinue = targetWritable.write(chunk, encoding, callback);
if (!canContinue) {
self.metrics.backpressurePauses++;
targetWritable.once('drain', () => {
// Отримано сигнал drain
});
}
},
final(callback) {
targetWritable.end(callback);
},
});
return monitored;
}
/**
* Виконує наскрізний конвеєр обробки логів із GZIP-компресією.
*/
public async executePipeline(
sourceStream: Readable,
destinationStream: Writable,
signal?: AbortSignal
): Promise<ProcessorMetrics> {
const sanitizer = this.createSanitizerTransform();
const gzip = createGzip({ level: 6 });
const monitoredTarget = this.createMetricsMonitor(destinationStream);
try {
await pipeline(
sourceStream,
sanitizer,
gzip,
monitoredTarget,
{ signal }
);
return { ...this.metrics };
} catch (error) {
if ((error as { name?: string }).name === 'AbortError') {
throw new Error('Log processing pipeline aborted by signal');
}
throw error;
}
}
public getMetrics(): ProcessorMetrics {
return { ...this.metrics };
}
}Виклик stream.write(chunk) у циклі з ігноруванням значення false призводить до вичерпання пам'яті. Коли внутрішня черга переповнена, дані продовжують безконтрольно накопичуватися в оперативній пам'яті процесу. Під навантаженням це призводить до аварійної зупинки сервера через Out Of Memory.
Завжди зупиняйте запис, коли stream.write() повертає false, та очікуйте події 'drain' (await once(stream, 'drain')) перед надсиланням наступного чанка.
Підписка на stream.on('data', handler) переводить потік у "текучий режим" (flowing mode), відключаючи зворотний тиск. Одночасна спроба ітерувати потік через for await (const chunk of stream) викликає втрату чанків, стани гонитви та порушення логіки.
Ніколи не поєднуйте різні парадигми читання: використовуйте або for await ... of, або pipeline(), або ручне читання через stream.on('readable').
Якщо в кастомному потоці Transform не викликати callback() в одній із логічних гілок, весь потік переходить у стан нескінченного очікування. Node.js не передаватиме наступні чанки і не завершить роботу. Конвеєр зависає назавжди.
Завжди викликайте callback(err) або callback(null, output) у кожній гілці коду, огортаючи логіку в блок try ... catch.
Використання readStream.pipe(transform).pipe(writeStream) у веб-сервері викликає витік ресурсів. Якщо клієнт раптово обриває TCP-з'єднання, потік запису видає помилку, проте pipe() залишає readStream відкритим. Файловий дескриптор блокується, поки система не поверне помилку EMFILE.
Завжди використовуйте stream.promises.pipeline(), яка гарантує коректне знищення всіх потоків ланцюжка при збої.
Джерело викликає writable.write(chunk), надсилаючи буфер або об'єкт.
Рантайм перевіряє чергу: якщо поріг перевищено, метод write() повертає false.
Readable потік призупиняє операції системного читання через виклик pause().
Контролер диска або мережевий сокет передає накопичені дані фізичному адресату.
Writable генерує подію drain, джерело викликає resume() і продовжує потік.