Потоки

Цель

Рассмотреть использоваия потоков в Node.js

Предварительныетребования

Знание основ JavaScript.

Установить Node.js и npm

Потоки это

Поток (_en.stream) - абстрактная последовательность инструкций или данных, привязанная к соответствующему имени потока.

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

Поддержка потоков включена в большинство языков программирования и современных операционных систем.

При запуске процесса ему предоставляются предопределённые стандартные потоки.

Возможность перенаправления потоков позволяет связывать различные программы, и придаёт системе гибкость, являющуюся частью философии Unix.

Потоки в Node.js

Если мы работаем с довольно большим файлом, то node будет буферезировать весь файл при каждом запросе, вследствии чего программа начнемт потреблять большое количество памяти, особенно при большом количестве подключенных пользователей с медленными каналами связи и при этом пользователю придется ждать пока весь файл не будет считан в память на сервере перед отправкой. Более оптимальным способом для чтения, записи а также для отправки на клиент файлов будет использование потоков. При использовании потоков файл доставляется по частям по мере чтения его с диска.

Существует 4 вида потоков:

• на чтение (readable) - потоки, из которых можно считывать данные
• на запись (writeable) - потоки, в которые могут быть записаны данные
• дуплексные (duplex) - потоки, которые являются читаемыми и записываемыми
• трансформирующие (transform) - дуплексные потоки, которые могут изменять или преобразовывать данные по мере их записи и чтения

Отметим, что при работе с сервером, параметры (req, res) являются потоками.

Любой из потоков может использовать метод .pipe(). Метод readable.pipe() присоединяет записываемый поток к читаемому, заставляя Readable поток автоматически выталкивать все свои данные в прикрепленный Writable. Поток данных при этом автоматически регулируется, таким образом, чтобы поток Writable не был перегружен более быстрым потоком Readable.

Модуль stream подключается с помощью команды require( 'stream' ).

Всем пользователям Node.js важно понимать, как работают стримы, и модуль stream сам по себе является самым удобным для разработчиков, которые создают новые типы экземпляров стримов. Разработчики, которые изначально изучили объекты стримов, редко будут нуждаться (если вообще будут) в прямом использовании модуляstream.

Потоки чтения (readable)

Все Readable потоки реализуют интерфейс, определенный классом stream.Readable.

Readable потоки эффективно работают в одном из двух режимов: flowing и paused.

В flowing (потоковом) режиме данные считываются из базовой системы автоматически и предоставляются в приложение как можно быстрее, используя события через интерфейс EventEmitter.

В paused(приостановленном) режиме метод stream.read() должен быть явно вызван для чтения фрагментов данных из потока.

Изначально для Readable потоков установлен paused режим. Для того чтобы перейти в потоковый режим, можно выполнить следующие действия:

• Добавить обработчик события 'data';
• Вызвать метод stream.resume();
• Вызвать метод stream.pipe() для передачи данных в поток записи

Для обратно перехода , используйте один из следующих способов:

• Если не привязки к другому потоку с помощью метода pipe(), вызовите метод stream.pause();
• Если есть, то для их удаления используйте метод stream.unpipe()

Для создания потока чтения используется метод fs.createReadStream(path[, options]). Для его использования необходимо подключить модуль 'fs'

Рассмотрим небольшой пример, допустим нас есть файл test.html, который бы мы хотели показать на странице.

var http = require('http');
var fs = require('fs');

http.createServer(function (req, res) {
    var stream = fs.createReadStream(__dirname + '/test.html'); // Создаем поток чтения файла
    stream.pipe(res); //выводим на страницу считанные данные
}).listen(8000);

Если у нас нет готового файла и мы просто хотим создать пустой поток Readable и записать в него информацию, то нам потребуется подключить модуль 'stream'.

Добавление данных в поток реализуется с помощью метода readable.push(chunk[, encoding]), в который передаются данные.

Пример:

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

rs.push('Hello, world\n');
rs.push('Goodbye');
rs.push(null);
rs.pipe(process.stdout); //вывод текста в консоль

Тут rs.push(null) сообщает потребителю, что rs закончил вывод данных.

Когда вы посылаете с помощью .push() данные в поток на чтение, они буферизируются до тех пор пока потребитель не будет готов их прочитать.

Тем не менее, в большинстве случаев будет лучше если мы не будем их буферизировать совсем, вместо этого будем генерировать их только когда данные запрашиваются потребителем.

Мы можем посылать данные кусками, определив функцию readable._read(size):

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

rs._read = function () {
    rs.push('Hello, world\n');
    rs.push('Goodbye');
    rs.push(null);
};
rs.pipe(process.stdout); //вывод текста в консоль

Также важно отметить что при открытии потока чтения, когда данные становятся доступными, возникает событие 'readable', и вы можете вызвать readable.read([size]) чтобы получить следующую порцию данных из буфера.

Когда поток завершится, .read() вернет null, потому что не останется доступных для чтения байтов.

process.stdin.on('readable', function () {
    var buf = process.stdin.read();
    console.dir(buf);
});

Потоки записи (writable)

Все Writable потоки реализуют интерфейс, определенный классом stream.Writable.

Хотя конкретные экземпляры потоков Writable могут различаться различными способами, все записываемые потоки следуют одному и тому же основному шаблону использования, как показано в примере ниже:

const myStream = getWritableStreamSomehow();
myStream.write('some data');
myStream.write('some more data');
myStream.end('done writing data');

Чтобы передать данные в поток на запись - вызовите writable.write(chunk[, encoding][, callback]), где chunk это набор данных которые вы хотите записать.

Если вы хотите сообщить что вы закончили запись - вызовите writable.end([chunk][, encoding][, callback]).

Для прерывания записи используют метод writable.destroy([error]).

Трансформирующий поток

Transform потоки это частный случай Duplex потоков (в обоих случаях они могут использоваться как для записи, так и чтения). Разница в том, что в случае трансформации отдаваемые данные так или иначе зависят от того что подается на вход.

Примеры включают потоки zlib или crypto потоки, которые сжимают, шифруют или дешифруют данные.

Также можно создать свой трансформирующий поток, как экземпляр класса Transform.

Например:

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

const upperCaseTr = new Transform({
    transform(chunk, encoding, callback) {
        this.push(chunk.toString().toUpperCase());
        callback();
    }
});
process.stdin.pipe(upperCaseTr).pipe(process.stdout);

В этом примере мы получаем с консоли текст, меняем ему регистр и записываем обратно в консоль.

process

process.stdin

Поток на чтение содержит стандартный системный поток ввода для вашей программы.

Если process.stdin указывает на терминал(проверяется вызовом tty.isatty()), тогда входящие данные будут буферизироваться построчно. Вы можете выключить построчную буферизацию вызвав process.stdin.setRawMode(true). Однако, имейте ввиду что в этом случае обработчики системных нажатий (таких как ^C и ^D) будут удалены.

process.stdout

Поток на запись, содержащий стандартный системный вывод для вашей программы. Посылайте туда данные, если вам нужно передать их в stdout.

process.stderr

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

Практическое задание

Создать два потока чтения и записи на основе файлов read.txt и write.txt. Необходимо переместить данные с файла для чтения в файл записи, так, чтобы записались только слова, в которых встречается буква 'a'.

Глоссарий

Ссылки