Потоки данных (Streams): Readable, Writable, Transform и Backpressure

Архитектурная база#

Архитектура Node.js спроектирована для обработки сетевого ввода-вывода с минимальным использованием вычислительных ресурсов. В стандартной модели монолитной буферизации (например, при вызове fs.readFile) файл целиком загружается в оперативную память процесса. Если сервер одновременно обрабатывает 100 запросов на выгрузку архивов по 500 МБ, приложению потребуется 50 ГБ RAM, что неминуемо приведет к падению процесса с ошибкой JavaScript heap out of memory или системному вызову OOM Killer ядра ОС.

Потоки данных (Streams) реализуют фундаментальный принцип потоковой обработки: разбиение непрерывного потока байтов на дискретные фрагменты (чанки — chunks) и их последовательную обработку по мере поступления. Потребление памяти процессом ограничивается лишь размером одного или нескольких чанков, находящихся в обработке (порядка нескольких десятков килобайт), независимо от общего объема передаваемого файла или времени жизни сетевого соединения.

В основе всех потоков Node.js лежит модуль node:stream, базирующийся на паттерне EventEmitter и внутренней структуре связанного списка буферов BufferList. В среде исполнения выделяются четыре базовых абстрактных класса:

1. Readable: источник данных, из которого можно читать байты (файловый дескриптор на чтение, входящее тело HTTP-запроса http.IncomingMessage, сетевой TCP-сокет).
2. Writable: приемник данных, в который можно записывать байты (файловый дескриптор на запись, исходящий ответ сервера http.ServerResponse, стандартный вывод process.stdout).
3. Duplex: двунаправленный поток, объединяющий независимые каналы чтения и записи (например, двусторонний TCP-сокет net.Socket).
4. Transform: специализированный подкласс Duplex, в котором данные на выходе математически или структурно вычисляются из входных данных (например, потоковое сжатие zlib.createGzip(), потоковое шифрование crypto.createCipheriv() или синтаксические анализаторы).

Внутренние буферы и highWaterMark#

Каждый поток инкапсулирует внутреннюю структуру состояния: _readableState для потоков чтения и _writableState для потоков записи. Центральным параметром конфигурации является highWaterMark:

• Для бинарных потоков значение по умолчанию составляет 64 КБ (65 536 байт), для файловых потоков fs.createReadStream — 64 КБ.
• Для потоков в режиме объектов (objectMode: true) значение по умолчанию равно 16 объектам.

highWaterMark не является жестким лимитом емкости памяти. Это сигнальный порог (threshold). Если приемник данных медленнее источника, данные продолжат накапливаться в оперативной памяти сверх highWaterMark, пока приложение не исчерпает доступную физическую RAM.

Физика явления Backpressure (Обратное давление)#

Явление Backpressure возникает в любой распределенной или асинхронной системе, когда скорость генерации данных источником превышает скорость их потребления или пропускную способность канала записи:

Скорость генерации (Producer)  >>  Скорость потребления (Consumer)
(например: диск NVMe 3 ГБ/с)        (например: сокет медленного 3G-клиента 50 КБ/с)

Если не управлять этой разницей скоростей, память сервера будет лавинообразно наполняться чанками, ожидающими отправки в сетевой сокет. Физика управления Backpressure в Node.js работает через двухфазный протокол:

1. Фаза насыщения: Метод writable.write(chunk) возвращает логическое значение. Если размер внутреннего буфера приемника превышает highWaterMark, write() возвращает false. Это строгое системное требование к источнику: немедленно приостановить чтение (readable.pause()).
2. Фаза дренажа: Когда нижележащий системный сокет или диск сбрасывает байты в буфер ядра ОС, внутренний буфер Writable-потока опустошается. Как только он освобождается, поток генерирует системное событие 'drain'. Получив 'drain', источник возобновляет чтение (readable.resume()).

Почему stream.pipe() устарел и опасен#

Метод readable.pipe(writable) исторически использовался для связывания потоков, однако в промышленной разработке он считается устаревшим (anti-pattern) по следующим критическим причинам:

• Утечка ресурсов при сбоях: Метод pipe() не распространяет ошибки по цепочке. Если промежуточный поток или приемник выбрасывает ошибку 'error', соединение не закрывается корректно. Файловые дескрипторы остаются заблокированными, что приводит к исчерпанию системных лимитов EMFILE: too many open files.
• Отсутствие поддержки отмены: pipe() не интегрирован со спецификацией AbortController и не может быть безопасно прерван клиентом без утечки буферов.
• Ручная очистка: Разработчик вынужден вручную вешать слушатели 'error', 'close' и 'finish' на каждое промежуточное звено пайплайна.

Современным промышленным стандартом является функция stream.promises.pipeline, которая гарантирует транзакционную надежность: при ошибке в любом звене цепочки все остальные потоки немедленно уничтожаются (destroy(err)), освобождая системные дескрипторы и сбрасывая ссылки для сборщика мусора.

Ментальная модель#

Работу потоков с обратным давлением можно представить как гидродинамический шлюз с управляемым клапаном.

[ 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() ] <--- Источник ВОЗОБНОВЛЯЕТ чтение!

На практике применяется#

• Потоковая обработка аналитических дампов (Big Data ETL): Чтение и разбор сотен гигабайт логов, транзакций в CSV или NDJSON построчно с мгновенной вставкой в базу данных батчами без утилизации всей памяти сервера.
• Проксирование и потоковое шифрование: Прием входящего multipart-запроса от клиента, шифрование чанков алгоритмом AES-256-GCM на лету и прямая пересылка в объектное хранилище (AWS S3, MinIO) через потоковый HTTP PUT.
• Сжатие трафика в реальном времени: Динамическое сжатие исходящих ответов API алгоритмами Gzip или Brotli с адаптацией под пропускную способность клиентского канала через zlib.createBrotliCompress().
• Генерация динамических отчетов на лету: Трансляция тяжелых XLSX/CSV выгрузок из PostgreSQL напрямую в HTTP-ответ клиенту без предварительного сохранения временного файла на диск.

Как это работает#

Реализация потокового Transform-анализатора: CSV в NDJSON#

Рассмотрим реализацию промышленного потока Transform, который принимает поток байтов CSV, корректно обрабатывает разрывы строк внутри чанков (chunk boundary issue) и эмитит валидные JSON-строки.

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)));
    }
  }
}

Асинхронная итерация (for await ... of) и AbortController#

Начиная с современных версий Node.js, стандартным и наиболее декларативным способом чтения Readable потоков стала конструкция for await (const chunk of stream). Движок Node.js автоматически управляет вызовами pause() и resume(), транслируя backpressure синхронно с телом цикла.

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;
    }
  }
}

Сравнение механизмов связывания потоков#

Готовый артефакт: HighThroughputLogProcessor#

Промышленный потоковый процессор журналов, объединяющий контроль обратного давления, ротацию чанков, подсчет контрольных метрик и корректное завершение работы (Graceful Shutdown).

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++;

        // Фильтрация некритичных логов в режиме экономии ресурсов
        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 (Newline Delimited JSON) с байтовым переводом строки
        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', () => {
            // Возобновление при дренаже
          });
        }
      },
      final(callback) {
        targetWritable.end(callback);
      },
    });

    return monitored;
  }

  /**
   * Запускает отказоустойчивую обработку логов из произвольного Readable в Writable с архивацией 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 };
  }
}

Распространённые ошибки#