Что такое Stream в Node.JS?

Stream в Node.js: эффективная обработка больших данных

Stream — это объект в Node.js, который представляет поток данных, которые можно читать, писать или трансформировать по частям (chunks), вместо загрузки всех данных в память одновременно. Это фундаментальная концепция для работы с файлами, сетевыми запросами и больших объемов данных.

Проблема без Streams

Представьте, что нужно прочитать файл размером 10 GB:

// ❌ Плохо — загружает весь файл в память
const fs = require('fs');
const data = fs.readFileSync('huge-file.txt', 'utf-8');
// Node.js использует 10 GB оперативной памяти!
// Очень медленно, может привести к краху

С большим файлом это приводит к:

• Истощению оперативной памяти
• Долгому времени на загрузку
• Блокировке других операций

Решение: Streams

// ✅ Хорошо — читает по частям (chunks)
const fs = require('fs');
const stream = fs.createReadStream('huge-file.txt');

stream.on('data', (chunk) => {
  // Обрабатываем по 64KB за раз (по умолчанию)
  console.log(`Получили chunk размером ${chunk.length} байт`);
});

stream.on('end', () => {
  console.log('Файл полностью обработан');
});

Типы Streams

1. Readable Stream (чтение)

Источник данных, из которого можно читать:

const fs = require('fs');

// Способ 1: обработчик 'data'
const readStream = fs.createReadStream('input.txt');
readStream.on('data', (chunk) => {
  console.log('Получен chunk:', chunk.toString());
});

// Способ 2: методы read()
const stream = fs.createReadStream('input.txt');
let chunk;
while ((chunk = stream.read()) !== null) {
  console.log('Read:', chunk);
}

// Способ 3: pipe
const readStream = fs.createReadStream('input.txt');
readStream.pipe(process.stdout); // выводит в консоль

2. Writable Stream (запись)

Время назначения, в которое можно писать:

const fs = require('fs');

const writeStream = fs.createWriteStream('output.txt');

writeStream.write('Строка 1\n');
writeStream.write('Строка 2\n');
writeStream.write('Строка 3\n');

writeStream.end(); // сигнал о завершении

writeStream.on('finish', () => {
  console.log('Запись завершена');
});

3. Transform Stream (трансформация)

Модифицирует данные по мере их прохождения:

const fs = require('fs');
const zlib = require('zlib');

// Встроенный Transform Stream для сжатия
const readStream = fs.createReadStream('large.txt');
const writeStream = fs.createWriteStream('large.txt.gz');
const gzip = zlib.createGzip();

// Пайпируем: читаем -> сжимаем -> пишем
readStream.pipe(gzip).pipe(writeStream);

writeStream.on('finish', () => {
  console.log('Файл сжат');
});

4. Duplex Stream (двусторонний)

Одновременно читает и пишет:

const net = require('net');

// TCP сокет — это Duplex Stream
const socket = net.createConnection({ port: 3000 });

// Читаем данные с сервера
socket.on('data', (chunk) => {
  console.log('Получили:', chunk.toString());
});

// Пишем данные на сервер
socket.write('Hello Server!');

Концепция Pipe (конвейер)

Pipe соединяет streams для обработки данных цепочкой:

const fs = require('fs');
const zlib = require('zlib');

// Читаем большой файл, сжимаем и пишем результат
fs.createReadStream('input.txt')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('input.txt.gz'))
  .on('finish', () => console.log('Done'));

// Это эквивалентно: cat input.txt | gzip > input.txt.gz

Практический пример: потоковая передача файла по HTTP

const express = require('express');
const fs = require('fs');
const path = require('path');

const app = express();

// Скачивание большого файла без загрузки в память
app.get('/download', (req, res) => {
  const filePath = path.join(__dirname, 'large-file.zip');
  
  res.setHeader('Content-Type', 'application/zip');
  res.setHeader('Content-Disposition', 'attachment; filename="large-file.zip"');
  
  // Пайпируем файл прямо в response
  fs.createReadStream(filePath).pipe(res);
});

app.listen(3000);

Обработка ошибок в Streams

const fs = require('fs');

const readStream = fs.createReadStream('input.txt');
const writeStream = fs.createWriteStream('output.txt');

// Обработка ошибок для читаемого stream
readStream.on('error', (error) => {
  if (error.code === 'ENOENT') {
    console.log('Файл не найден');
  } else {
    console.error('Ошибка при чтении:', error);
  }
});

// Обработка ошибок для writeable stream
writeStream.on('error', (error) => {
  console.error('Ошибка при записи:', error);
});

// Pipe с обработкой ошибок
readStream
  .pipe(writeStream)
  .on('error', (error) => {
    console.error('Ошибка в pipe:', error);
  });

Backpressure (обратное давление)

Важная концепция для балансировки скоростей чтения и записи:

const fs = require('fs');

const readStream = fs.createReadStream('large.txt');
const writeStream = fs.createWriteStream('output.txt');

// ❌ Неправильно — может привести к утечке памяти
readStream.on('data', (chunk) => {
  writeStream.write(chunk);
  // Если писать медленнее, чем читать — буфер растёт
});

// ✅ Правильно — обработка backpressure
readStream.on('data', (chunk) => {
  const ok = writeStream.write(chunk);
  
  if (!ok) {
    // Буфер заполнен, паузируем чтение
    readStream.pause();
  }
});

writeStream.on('drain', () => {
  // Буфер очищен, возобновляем чтение
  readStream.resume();
});

// Или просто используй pipe (обрабатывает backpressure автоматически)
readStream.pipe(writeStream);

Когда использовать Streams

• Большие файлы (более 50 MB)
• Потоковые данные (видео, аудио)
• HTTP запросы/ответы (скачивание, загрузка файлов)
• Логирование в файлы
• Реал-тайм обработка данных
• Интеграция с базами данных (экспорт/импорт больших данных)

Ключевые выводы

• Streams обрабатывают данные по частям, а не целиком в памяти
• Pipe — эффективный способ связать несколько streams
• Backpressure критична для стабильности
• Error handling — обязателен для production кода
• Streams — это основа высокопроизводительного Node.js