Потоки (Streams) — это механизм обработки данных небольшими частями, позволяющий эффективно работать с большими файлами без переполнения памяти.
Обзор типов потоков
| Тип | Направление | Описание |
|---|---|---|
Readable |
Только чтение | Источник, из которого считываются данные |
Writable |
Только запись | Приемник, в который записываются данные |
Transform |
Чтение и запись | Преобразует или модифицирует проходящие через него данные |
Duplex |
Чтение и запись | Поддерживает чтение и запись с независимыми каналами (например, TCP-сокет) |
Конвейер pipeline и модуль stream/promises
Современный Node.js предоставляет функцию pipeline для безопасного соединения потоков с автоматической обработкой ошибок и освобождением ресурсов:
const { pipeline } = require('stream/promises');
const fs = require('fs');
const zlib = require('zlib');
async function compressFile() {
await pipeline(
fs.createReadStream('source.tar'),
zlib.createGzip(),
fs.createWriteStream('source.tar.gz')
);
console.log('Сжатие завершено');
}
Создание собственных потоков
1. Пользовательский поток Readable
const { Readable } = require('stream');
function createTimestampStream() {
let count = 0;
return new Readable({
objectMode: true,
read() {
if (count >= 5) {
this.push(null); // Завершение потока
} else {
this.push({ id: ++count, time: new Date() });
}
}
});
}
2. Пользовательский поток Transform
const { Transform } = require('stream');
const uppercaseTransformer = new Transform({
transform(chunk, encoding, callback) {
const transformed = chunk.toString().toUpperCase();
callback(null, transformed);
}
});
3. Пользовательский поток Writable
const { Writable } = require('stream');
const consoleWriter = new Writable({
objectMode: true,
write(chunk, encoding, callback) {
console.log('Запись порции данных:', chunk);
callback();
}
});
Режим Flowing против Paused
Потоки для чтения работают либо в приостановленном (paused), либо в непрерывном (flowing) режиме:
const stream = fs.createReadStream('data.csv');
// Подключение слушателя 'data' автоматически включает режим flowing
stream.on('data', (chunk) => {
console.log(`Получено ${chunk.length} байт`);
});
// Явное управление потоком
stream.pause(); // Временная приостановка событий
stream.resume(); // Возобновление передачи данных
События потоков
| Событие | Применимо к | Условие возникновения |
|---|---|---|
data |
Readable |
Генерируется при готовности порции данных к считыванию |
end |
Readable |
Генерируется, когда доступные для чтения данные закончились |
finish |
Writable |
Генерируется, когда все записанные данные были сброшены в приемник |
error |
All | Генерируется при возникновении необработанной ошибки в потоке |
close |
All | Генерируется после закрытия базового дескриптора или ресурса |