Потоки: большие файлы по частям

readFile() загружает весь файл в память сразу. Для журнала сервера размером в несколько гигабайт это не годится: памяти не хватит. Поток (stream) читает или пишет данные небольшими частями — по 64 КБ — и обрабатывает их по мере поступления. Так можно работать с файлом любого размера.

Поток записи

createWriteStream() из модуля node:fs открывает файл для записи частями. Создадим журнал посещений блога на десять тысяч строк:

import { createWriteStream } from 'node:fs';
 
const log = createWriteStream('access.log');
 
for (let i = 1; i <= 10000; i++) {
  const page = i % 3 === 0 ? '/posts/1' : '/';
  log.write(`2025-01-15 GET ${page}\n`);
}
 
log.end(() => console.log('Журнал записан'));

write() добавляет кусок данных, end() закрывает файл и вызывает функцию, когда всё записано. Рядом появится файл access.log.

Поток чтения

createReadStream() отдаёт файл частями. Потоки поддерживают for await, поэтому части перебирают обычным циклом:

import { createReadStream } from 'node:fs';
import { writeFile } from 'node:fs/promises';
 
await writeFile('access.log', 'GET /\n'.repeat(50000));
 
let bytes = 0;
let chunks = 0;
for await (const chunk of createReadStream('access.log')) {
  bytes += chunk.length;
  chunks++;
}
 
console.log(`Прочитано ${bytes} байт за ${chunks} частей`);

Выведется «Прочитано 300000 байт за 5 частей»: каждая часть — объект Buffer размером до 64 КБ.

Чтение по строкам

Части режут файл в случайных местах, посреди строки. Чтобы получать целые строки, поток передают модулю node:readline:

import { createReadStream } from 'node:fs';
import { writeFile } from 'node:fs/promises';
import readline from 'node:readline';
 
await writeFile('access.log', 'GET /\nGET /posts/1\nGET /\nGET /posts/1\nGET /posts/1\n');
 
const lines = readline.createInterface({ input: createReadStream('access.log') });
 
let views = 0;
for await (const line of lines) {
  if (line.includes('/posts/1')) views++;
}
 
console.log(`Статью 1 открыли ${views} раза`); // 3 раза

В памяти в каждый момент находится только одна строка, поэтому код одинаково работает и для пяти строк, и для пяти миллионов.

pipeline: из потока в поток

Функция pipeline() из node:stream/promises соединяет потоки цепочкой: данные идут из первого в последний. Так файл сжимают в архив .gz модулем node:zlib:

import { createReadStream, createWriteStream } from 'node:fs';
import { stat, writeFile } from 'node:fs/promises';
import { pipeline } from 'node:stream/promises';
import { createGzip } from 'node:zlib';
 
await writeFile('access.log', 'GET /posts/1\n'.repeat(20000));
 
await pipeline(createReadStream('access.log'), createGzip(), createWriteStream('access.log.gz'));
 
const before = (await stat('access.log')).size;
const after = (await stat('access.log.gz')).size;
console.log(after < before); // true

pipeline() сама закрывает файлы и передаёт ошибку любого звена в await.

Частая ошибка

Соединять потоки старым методом .pipe() опасно: если поток чтения упадёт (например, файла нет), поток записи останется открытым, а ошибка может уронить программу. Используйте pipeline() и оборачивайте её в try...catch:

import { createReadStream, createWriteStream } from 'node:fs';
import { pipeline } from 'node:stream/promises';
 
try {
  await pipeline(createReadStream('missing.log'), createWriteStream('copy.log'));
} catch (error) {
  console.log(error.code); // ENOENT
}

Спасибо, что читаете!