Полное руководство по потокам в Node.js

Потоки в Node.js — это мощный способ непрерывного чтения и записи данных. Они позволяют эффективно обрабатывать данные, особенно при работе с большими объемами информации или операциями ввода-вывода. В этом руководстве будут рассмотрены типы потоков, способы их использования, а также практические примеры, которые помогут вам понять, как потоки работают в Node.js.

Что такое потоки?

Потоки — это объекты, которые позволяют непрерывно считывать данные из источника или записывать данные в пункт назначения. Они идеально подходят для обработки данных по частям, а не для одновременного чтения или записи целых файлов или буферов. Это особенно полезно при работе с большими наборами данных, поскольку позволяет значительно сократить использование памяти.

Ключевые понятия

• Режим передачи: данные автоматически передаются от источника к месту назначения.
• Режим паузы: данные необходимо вручную считать из источника.
• Читаемые потоки: потоки, из которых можно читать данные.
• Потоки с возможностью записи: потоки, в которые можно записывать данные.
• Дуплексные потоки: потоки, которые могут как читать, так и записывать данные.
• Потоки преобразования: потоки, которые изменяют или преобразуют данные при их чтении или записи.

Типы потоков

1. Чтимые потоки: эти потоки позволяют читать данные. Примеры включают fs.createReadStream() и http.IncomingMessage.

2. Потоки с возможностью записи: эти потоки позволяют записывать данные. Примеры включают fs.createWriteStream() и http.ServerResponse.

3. Дуплексные потоки: эти потоки могут читать и записывать данные. Примеры включают TCP-сокеты и net.Duplex.

4. Потоки преобразования: это тип дуплексного потока, который может изменять данные во время их чтения или записи. Примеры включают zlib.createGzip() для сжатия.

Создание читаемых потоков

Вы можете создать читаемый поток, используя встроенный модуль fs для чтения файлов или используя поток.Readable для создания пользовательских читаемых потоков.

Пример: чтение файла с помощью читаемого потока

const fs = require('fs');

// Create a readable stream
const readableStream = fs.createReadStream('example.txt', { encoding: 'utf8' });

// Handling the 'data' event
readableStream.on('data', (chunk) => {
    console.log('New chunk received:', chunk);
});

// Handling the 'end' event
readableStream.on('end', () => {
    console.log('No more data to read.');
});

Пример: пользовательский читаемый поток

const { Readable } = require('stream');

class MyReadableStream extends Readable {
    constructor(options) {
        super(options);
        this.current = 0;
    }

    _read(size) {
        if (this.current  {
    console.log('Received:', chunk.toString());
});

Создание записываемых потоков

Вы можете создавать записываемые потоки с помощью модуля fs или путем расширения классаstream.Writable.

Пример: запись в файл с помощью записываемого потока

const fs = require('fs');

// Create a writable stream
const writableStream = fs.createWriteStream('output.txt');

// Write data to the stream
writableStream.write('Hello, World!\n');
writableStream.write('Writing to a file using streams.\n');

// End the stream
writableStream.end(() => {
    console.log('Finished writing to file.');
});

Пример: пользовательский поток с возможностью записи

const { Writable } = require('stream');

class MyWritableStream extends Writable {
    _write(chunk, encoding, callback) {
        console.log('Writing:', chunk.toString());
        callback(); // Call when done
    }
}

const myWritableStream = new MyWritableStream();
myWritableStream.write('Hello, World!\n');
myWritableStream.write('Writing to custom writable stream.\n');
myWritableStream.end();

Использование дуплексных потоков

Дуплексные потоки могут одновременно читать и записывать данные. Распространенный вариант использования — TCP-сокеты.

Пример: создание дуплексного потока

const { Duplex } = require('stream');

class MyDuplexStream extends Duplex {
    _read(size) {
        this.push('Data from duplex stream\n');
        this.push(null); // No more data
    }

    _write(chunk, encoding, callback) {
        console.log('Received:', chunk.toString());
        callback();
    }
}

const myDuplexStream = new MyDuplexStream();
myDuplexStream.on('data', (chunk) => {
    console.log('Reading:', chunk.toString());
});

// Write to the duplex stream
myDuplexStream.write('Hello, Duplex!\n');
myDuplexStream.end();

Использование потоков преобразования

Потоки преобразования полезны для изменения данных, проходящих через поток. Например, вы можете использовать поток преобразования для сжатия данных.

Пример: создание потока преобразования

const { Transform } = require('stream');

class MyTransformStream extends Transform {
    _transform(chunk, encoding, callback) {
        const upperChunk = chunk.toString().toUpperCase();
        this.push(upperChunk);
        callback();
    }
}

const myTransformStream = new MyTransformStream();
myTransformStream.on('data', (chunk) => {
    console.log('Transformed:', chunk.toString());
});

// Pipe data through the transform stream
process.stdin.pipe(myTransformStream).pipe(process.stdout);

Трубопроводные потоки

Одной из мощных особенностей потоков является возможность соединять их вместе. Трубопровод позволяет соединить поток, доступный для чтения, с потоком, доступным для записи, что упрощает передачу данных.

Пример: трубопроводные потоки

const fs = require('fs');

// Create a readable stream
const readableStream = fs.createReadStream('input.txt');

// Create a writable stream
const writableStream = fs.createWriteStream('output.txt');

// Pipe the readable stream to the writable stream
readableStream.pipe(writableStream);

writableStream.on('finish', () => {
    console.log('Data has been written to output.txt');
});

Потоковая передача событий в Node.js

1. Читаемые потоковые события

Читаемые потоки отправляют несколько важных событий, которые помогут вам управлять потоком данных:

• данные: генерируется, когда фрагмент данных доступен для чтения.
• end: генерируется, когда больше нет данных для чтения.
• ошибка: генерируется, когда во время чтения возникает ошибка.
• close: генерируется, когда поток и любые базовые ресурсы (например, файловые дескрипторы) были закрыты.

Пример: читаемые события потока

const fs = require('fs');

const readableStream = fs.createReadStream('example.txt');

readableStream.on('data', (chunk) => {
    console.log('Received chunk:', chunk.toString());
});

readableStream.on('end', () => {
    console.log('No more data to read.');
});

readableStream.on('error', (err) => {
    console.error('Error occurred:', err);
});

readableStream.on('close', () => {
    console.log('Stream closed.');
});

2. События записываемых потоков

Потоки, доступные для записи, также генерируют несколько событий:

• drain: генерируется, когда поток готов принять больше данных после заполнения.
• finish: генерируется, когда все данные записаны в поток и вызван метод end().
• ошибка: генерируется, когда во время записи возникает ошибка.
• close: генерируется, когда поток и все базовые ресурсы закрыты.

Пример: записываемые события потока

const fs = require('fs');

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

writableStream.on('finish', () => {
    console.log('All data has been written to output.txt');
});

writableStream.on('error', (err) => {
    console.error('Error occurred:', err);
});

// Writing data
writableStream.write('Hello, World!\n');
writableStream.write('Writing to a file using streams.\n');
writableStream.end(); // Call end to finish the writing process

3. Преобразование событий потоков

Потоки преобразования наследуют события как от читаемых, так и от записываемых потоков и выдают:

• данные: генерируется, когда преобразованный фрагмент доступен для чтения.
• end: генерируется, когда больше нет данных для преобразования.
• ошибка: генерируется, когда во время преобразования возникает ошибка.
• finish: генерируется, когда все данные обработаны и записаны.

Пример: преобразование событий потока

const { Transform } = require('stream');

class MyTransformStream extends Transform {
    _transform(chunk, encoding, callback) {
        const upperChunk = chunk.toString().toUpperCase();
        this.push(upperChunk);
        callback();
    }
}

const myTransformStream = new MyTransformStream();

myTransformStream.on('data', (chunk) => {
    console.log('Transformed chunk:', chunk.toString());
});

myTransformStream.on('end', () => {
    console.log('No more data to transform.');
});

myTransformStream.on('error', (err) => {
    console.error('Error occurred:', err);
});

// Write data to the transform stream
myTransformStream.write('Hello, World!\n');
myTransformStream.write('Transforming this text.\n');
myTransformStream.end(); // End the stream

Краткое изложение событий

• Чтимые потоки: данные, конец, ошибка, закрытие
• Потоки с возможностью записи: слив, завершение, ошибка, закрытие
• Преобразование потоков: наследует данные, конец, ошибку, завершение как из читаемых, так и из записываемых потоков

Заключение

Потоки в Node.js предоставляют мощный и эффективный способ непрерывной обработки данных. Они позволяют читать и записывать данные по частям, что делает их особенно полезными для операций ввода-вывода и работы с большими наборами данных. Понимание того, как создавать и использовать различные типы потоков, а также как обрабатывать события, поможет вам создавать более эффективные и масштабируемые приложения на Node.js.

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