Node.js Stream
Node.js Stream — это абстрактный интерфейс для обработки потоковых данных, широко используемый в таких сценариях, как файловые операции и сетевое взаимодействие.
Основное преимущество потоковой обработки данных заключается в том, что данные можно обрабатывать во время передачи, не дожидаясь загрузки всех данных, что позволяет Node.js эффективно обрабатывать большие объёмы данных, не потребляя чрезмерный объём памяти.
Stream — это абстрактный интерфейс, и многие объекты в Node реализуют этот интерфейс. Например, объект request для HTTP-запроса к серверу является Stream, как и stdout (стандартный вывод).
Node.js Stream — это способ обработки потоков данных. Он позволяет обрабатывать данные потоковым способом вместо загрузки всех данных в память сразу. Это очень полезно для обработки больших объёмов данных или обеспечения эффективной передачи данных.
В Node.js существует четыре типа потоков:
Readable- операции чтения.
Writable- операции записи.
Duplex- операции чтения и записи.
Transform- операции, которые записывают данные, а затем читают результат.
Все объекты Stream являются экземплярами EventEmitter. К типичным событиям относятся:
data- Срабатывает, когда данные доступны для чтения.
end- Срабатывает, когда больше нет данных для чтения.
error- Срабатывает при возникновении ошибки во время получения или записи.
finish- Срабатывает, когда все данные были записаны в нижележащую систему.
В этом руководстве будут представлены распространённые операции с потоками.
Чтение данных из потока
К распространённым читаемым потокам относятся потоки чтения файлов и потоки ответов на сетевые запросы.
Создайте файл input.txt со следующим содержимым:
Example官网地址：www.example.com
Создайте файл main.js со следующим кодом:
Пример
var data = '';
// Создать читаемый поток
var readerStream = fs.createReadStream('input.txt');
// Установить кодировку utf8.
readerStream.setEncoding('UTF8');
// Обработка событий потока --> data, end и error
readerStream.on('data', function(chunk) {
data += chunk;
});
readerStream.on('end',function(){
console.log(data);
});
readerStream.on('error', function(err){
console.log(err.stack);
});
console.log("Выполнение программы завершено");
Результат выполнения приведённого выше кода следующий:
程序执行完毕 Example官网地址：www.example.com
Поток для записи
Поток для записи используется для записи данных в пункт назначения. К распространённым потокам для записи относятся потоки записи в файл и потоки отправки сетевых запросов.
Создайте файл main.js со следующим кодом:
Пример
var data = 'Пример официального адреса сайта: www.example.com';
// Создать поток для записи, выполняющий запись в файл output.txt
var writerStream = fs.createWriteStream('output.txt');
// Записать данные с использованием кодировки utf8
writerStream.write(data,'UTF8');
// Обозначить конец файла
writerStream.end();
// Обработать события потока --> finish, error
writerStream.on('finish', function() {
console.log("Запись завершена.");
});
writerStream.on('error', function(err){
console.log(err.stack);
});
console.log("Выполнение программы завершено");
Приведённая выше программа записывает данные из переменной data в файл output.txt. Результат выполнения кода выглядит следующим образом:
$ node main.js 程序执行完毕 写入完成。
Просмотрите содержимое файла output.txt:
$ cat output.txt Example官网地址：www.example.com
Дуплексный поток
Дуплексный поток обладает возможностями чтения и записи.
Типичный дуплексный поток — это TCP-сокет.
Пример
// Создать TCP-сервер
const server = net.createServer((socket) => {
console.log('Клиент подключен.');
// Чтение данных клиента
socket.on('data', (data) => {
console.log('Полученные данные:', data.toString());
});
// Отправка данных клиенту
socket.write('Привет, клиент!\n');
// Прослушивать событие close
socket.on('end', () => {
console.log('Клиент отключен.');
});
});
server.listen(3000, () => {
console.log('Сервер прослушивает порт 3000.');
});
Поток преобразования
Поток преобразования — это особый тип дуплексного потока, который может изменять или преобразовывать данные. К распространённым потокам преобразования относятся потоки сжатия и распаковки.
Пример
const fs = require('fs');
// Создать читаемый поток
const readableStream = fs.createReadStream('example.txt');
// Создать поток преобразования (сжатие)
const gzip = zlib.createGzip();
// Создать поток для записи
const writableStream = fs.createWriteStream('example.txt.gz');
// Направить читаемый поток в поток преобразования, затем направить в поток для записи
readableStream.pipe(gzip).pipe(writableStream);
// Прослушивать событие завершения
writableStream.on('finish', () => {
console.log('Файл успешно сжат.');
});
Поток конвейера
Конвейер предоставляет механизм для соединения выходного потока с входным потоком.
Обычно мы используем его, чтобы получать данные из одного потока и передавать данные в другой поток.
Как показано на изображении выше, мы сравниваем файлы с бочками, наполненными водой, а вода — это содержимое файлов. Мы используем трубу, чтобы соединить две бочки, так что вода перетекает из одной бочки в другую, что постепенно реализует процесс копирования больших файлов.
В следующем примере мы читаем содержимое одного файла и записываем содержимое в другой файл.
Задайте содержимое файла input.txt следующим образом:
Example官网地址：www.example.com 管道流操作实例
Создайте файл main.js со следующим кодом:
Пример
// Создать читаемый поток
var readerStream = fs.createReadStream('input.txt');
// Создать поток для записи
var writerStream = fs.createWriteStream('output.txt');
// Операции чтения и записи через канал
// Прочитать содержимое input.txt и записать содержимое в output.txt
readerStream.pipe(writerStream);
console.log("Выполнение программы завершено");
Результат выполнения кода выглядит следующим образом:
$ node main.js 程序执行完毕
Просмотрите содержимое файла output.txt:
$ cat output.txt Example官网地址：www.example.com 管道流操作实例
Цепочка потоков
Связывание в цепочку — это механизм подключения выходного потока к другому потоку и создания цепочки из нескольких потоковых операций.
Сцепленные потоки обычно используются для операций через каналы.
Далее мы будем использовать каналы и цепочки для сжатия и распаковки файлов.
Создайте файл compress.js со следующим кодом:
Пример
var zlib = require('zlib');
// Сжать input.txt в input.txt.gz
fs.createReadStream('input.txt')
.pipe(zlib.createGzip())
.pipe(fs.createWriteStream('input.txt.gz'));
console.log("Сжатие файла завершено.");
Результат выполнения кода выглядит следующим образом:
$ node compress.js 文件压缩完成。
После выполнения вышеуказанных операций мы можем увидеть, что в текущем каталоге был создан сжатый файл input.txt.gz.
Теперь давайте распакуем этот файл. Создайте файл decompress.js со следующим кодом:
Пример
var zlib = require('zlib');
// Распаковать input.txt.gz в input.txt
fs.createReadStream('input.txt.gz')
.pipe(zlib.createGunzip())
.pipe(fs.createWriteStream('input.txt'));
console.log("Распаковка файла завершена.");
Результат выполнения кода выглядит следующим образом:
$ node decompress.js 文件解压完成。
Приостановка и возобновление
Поток для чтения может приостанавливать и возобновлять чтение данных.
Пример
const readableStream = fs.createReadStream('example.txt', 'utf8');
readableStream.on('data', (chunk) => {
console.log('Получен фрагмент:', chunk);
readableStream.pause(); // Приостановить чтение
setTimeout(() => {
readableStream.resume(); // Возобновить чтение
}, 1000);
});
Уничтожение
Вы можете уничтожить поток, чтобы освободить ресурсы.
Пример
const readableStream = fs.createReadStream('example.txt', 'utf8');
readableStream.on('data', (chunk) => {
console.log('Получен фрагмент:', chunk);
readableStream.destroy(); // Уничтожить поток
});