Что такое Streams в Node.js и какие типы стримов существуют?

Что такое Streams в Node.js и какие типы стримов существуют?

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

Зачем нужны Streams?

Без Streams:

const data = fs.readFileSync('huge.txt');
// Out of Memory на 100 GB файле

С Streams:

const stream = fs.createReadStream('huge.txt');
stream.on('data', (chunk) => {
  console.log(`Chunk ${chunk.length} bytes`);
});

4 типа Streams

1. Readable Stream (читаемый)

const readStream = fs.createReadStream('input.txt', {
  encoding: 'utf-8',
  highWaterMark: 16 * 1024
});

readStream.on('data', (chunk) => {
  console.log(`Получен ${chunk.length} bytes`);
});

readStream.on('end', () => {
  console.log('Файл прочитан');
});

readStream.pause();  // Приостановить
readStream.resume(); // Возобновить

2. Writable Stream (записываемый)

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

writeStream.write('Строка 1\n');
writeStream.write('Строка 2\n');
writeStream.end('Последняя\n');

writeStream.on('finish', () => {
  console.log('Записано');
});

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

for (let i = 0; i < 1000000; i++) {
  const canWrite = writeStream.write(`Line ${i}\n`);
  
  if (!canWrite) {
    console.log('Буфер переполнен');
  }
}

writeStream.on('drain', () => {
  console.log('Буфер опустел');
});

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

const gzip = zlib.createGzip();

fs.createReadStream('input.txt')
  .pipe(gzip)
  .pipe(fs.createWriteStream('input.txt.gz'))
  .on('finish', () => {
    console.log('Архивировано');
  });

4. Transform Stream (трансформирующий)

const uppercaseTransform = new Transform({
  transform(chunk, encoding, callback) {
    const transformed = chunk.toString().toUpperCase();
    callback(null, transformed);
  }
});

fs.createReadStream('input.txt')
  .pipe(uppercaseTransform)
  .pipe(process.stdout);

Pipe — главный инструмент

// Вместо ручной обработки
readStream.on('data', (chunk) => {
  writeStream.write(chunk);
});

// Пишем так
readStream.pipe(writeStream);

// Цепочка
fs.createReadStream('input.txt')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('input.gz'))
  .on('finish', () => {
    console.log('Готово!');
  });

Практические примеры

Загрузка большого файла на сервер:

app.post('/upload', (req, res) => {
  const fileName = `upload_${Date.now()}.bin`;
  
  req
    .pipe(fs.createWriteStream(fileName))
    .on('finish', () => {
      res.send(`Загружено: ${fileName}`);
    })
    .on('error', (err) => {
      res.status(500).send(`Ошибка: ${err.message}`);
    });
});

Чтение БД большим объёмом:

const queryStream = db.query(
  'SELECT * FROM users WHERE created > $1',
  [date]
).stream();

queryStream
  .pipe(csv.format({ headers: true }))
  .pipe(fs.createWriteStream('users.csv'))
  .on('finish', () => {
    console.log('CSV экспортирован');
  });

Real-time логирование:

const timestampTransform = new Transform({
  transform(chunk, encoding, callback) {
    const log = `[${new Date().toISOString()}] ${chunk}`;
    callback(null, log);
  }
});

process.stdin
  .pipe(timestampTransform)
  .pipe(fs.createWriteStream('app.log', { flags: 'a' }));

Ключевые события

• data — Получен chunk
• end — Данные закончились
• error — Ошибка
• finish — Все записано
• drain — Буфер опустел
• pipe — Подключен через pipe()

Главное помните

1. Streams экономят память — по частям
2. 4 типа: Readable, Writable, Duplex, Transform
3. pipe() соединяет streams
4. backpressure — контроль скорости
5. Real-world: файлы, сеть, БД, архивация