worker_threads: настоящие потоки для CPU-задач

Когда тяжёлое вычисление блокирует событийный цикл, спасают worker_threads — настоящие потоки внутри процесса Node, у каждого свой движок V8 и своя память.

worker_threads — встроенный модуль Node.js, позволяющий запускать JavaScript в отдельных потоках операционной системы внутри одного процесса. Каждый поток получает собственный экземпляр V8 и собственный событийный цикл, а общаются потоки сообщениями.

Из раздела про асинхронность мы знаем главную особенность Node: код выполняется в одном потоке, а событийный цикл лишь переключается между готовыми задачами. Для операций ввода-вывода (файлы, сеть, БД) этого хватает с запасом — ожидание уходит в фон, поток свободен. Но как только появляется чистое вычисление — хеширование пароля, парсинг гигантского JSON, обработка изображения, перемножение матриц — событийный цикл встаёт колом: пока считается результат, ни один другой запрос не обслуживается. Именно эту дыру и закрывают worker_threads.

Зачем это на практике

Представьте веб-сервер, который иногда генерирует PDF или считает контрольную сумму большого файла. Если делать это в основном потоке, на время вычисления сервер перестаёт отвечать вообще всем клиентам — даже на лёгкий GET /health. Пользователи видят зависание, балансировщик считает узел мёртвым. Вынеся вычисление в worker, вы освобождаете событийный цикл: основной поток продолжает принимать соединения, а тяжёлая работа крутится параллельно на другом ядре. Это разница между «сервис тормозит под нагрузкой» и «сервис держит нагрузку ровно».

Как блокируется событийный цикл

Сначала покажем проблему. Синхронный тяжёлый цикл не даёт событийному циклу обработать ничего другого:

// blocking.js — НЕ делайте так в основном потоке сервера
function heavySum(n) {
  let s = 0;
  for (let i = 0; i < n; i++) s += Math.sqrt(i);
  return s;
}

console.log('старт');
const result = heavySum(5_000_000_00); // десятки секунд чистого CPU
console.log('готово', result);
// всё это время сервер в том же процессе НЕ отвечает ни на один запрос

Никакой async/await здесь не поможет: await уступает управление только на настоящих асинхронных операциях, а тут — сплошной синхронный счёт. Нужен отдельный поток.

Первый worker

Модуль worker_threads даёт класс Worker (создаёт поток) и набор примитивов внутри потока. Worker — это отдельный файл, который Node запускает в новом потоке. Общение идёт через события message.

// worker.js — выполняется в отдельном потоке
const { parentPort, workerData } = require('worker_threads');

function heavySum(n) {
  let s = 0;
  for (let i = 0; i < n; i++) s += Math.sqrt(i);
  return s;
}

const answer = heavySum(workerData.n); // считаем, не мешая главному потоку
parentPort.postMessage(answer);        // отдаём результат обратно

// main.js — основной поток
const { Worker } = require('worker_threads');

function runHeavy(n) {
  return new Promise((resolve, reject) => {
    const worker = new Worker('./worker.js', { workerData: { n } });
    worker.on('message', resolve);                 // пришёл результат
    worker.on('error', reject);                     // ошибка внутри потока
    worker.on('exit', (code) => {
      if (code !== 0) reject(new Error('worker exit ' + code));
    });
  });
}

console.log('сервер свободен и принимает запросы');
runHeavy(500_000_000).then((r) => console.log('итог из потока:', r));

Обратите внимание: workerData передаёт входные данные в поток при создании, а parentPort.postMessage возвращает результат. Главный поток в это время не блокирован — он спокойно обслуживает другие соединения. Обёртка в Promise превращает «потоковую» механику в привычный await.

Обмен сообщениями: MessagePort

Связь между потоками — это пара портов (MessagePort), как телефонная линия. В главном потоке вы держите объект worker (это и есть один конец канала), внутри потока — parentPort (другой конец). Сообщения двунаправленны, и поток может слать промежуточные результаты, а не только финал:

// внутри worker.js: слать прогресс
for (let part = 1; part <= 4; part++) {
  // ... посчитали четверть ...
  parentPort.postMessage({ type: 'progress', part });
}
parentPort.postMessage({ type: 'done', result: 42 });

// в main.js: разбирать по типу
worker.on('message', (msg) => {
  if (msg.type === 'progress') console.log('прогресс', msg.part, '/4');
  else if (msg.type === 'done') console.log('результат', msg.result);
});

Важно понимать модель памяти: у каждого потока своя куча. Переменные не разделяются — нельзя из главного потока «дотянуться» до объекта внутри worker. Всё, что пересекает границу, проходит через postMessage, и объекты при этом копируются алгоритмом structured clone (как при JSON, но богаче: переживают Map, Set, Date, ArrayBuffer). Функции и замыкания передать нельзя.

Передача буферов без копирования

Копирование больших данных туда-обратно само по себе стоит времени. Если вы передаёте крупный ArrayBuffer и он больше не нужен в источнике, его можно передать во владение (transfer), а не копировать: буфер «телепортируется» в другой поток, а в исходном становится недоступен.

const buf = new Uint8Array(50_000_000); // 50 МБ
// второй аргумент — список объектов, которые отдаются во владение
worker.postMessage(buf, [buf.buffer]);
// после этого buf.buffer в этом потоке пуст (byteLength === 0)

Это нулевое копирование: вместо пересылки 50 МБ переезжает только владение памятью. Приём незаменим для аудио, видео, изображений, бинарных протоколов.

SharedArrayBuffer кратко

Иногда нужна по-настоящему общая память, которую потоки видят одновременно. Её даёт SharedArrayBuffer: один и тот же блок байтов доступен из нескольких потоков сразу — без копий и без передачи владения.

// main.js
const shared = new SharedArrayBuffer(4);    // 4 байта = одно 32-битное число
const view = new Int32Array(shared);
worker.postMessage({ shared });             // делимся ссылкой на общий блок

// worker.js — пишет в ту же память
const { shared } = workerData;
const view = new Int32Array(shared);
Atomics.add(view, 0, 1);                     // атомарно увеличить общий счётчик

Поскольку память общая, обычная запись приводит к гонкам данных. Поэтому к SharedArrayBuffer прилагается объект Atomics — атомарные операции (add, load, store, wait, notify), которые гарантируют корректность при одновременном доступе. Это мощно, но и опасно: ровно те же классические проблемы многопоточности (гонки, дедлоки), от которых однопоточный Node нас обычно избавляет. Для большинства задач хватает postMessage с копией или transfer; SharedArrayBuffer берут, когда нужна высокочастотная синхронизация большого общего массива.

Как это работает под капотом

Worker — это не процесс, а поток ОС внутри того же процесса Node. Но V8 спроектирован так, что каждый изолят (изолированный экземпляр движка) живёт сам по себе: своя куча, свой сборщик мусора, свой стек. Поэтому потоки Node не делят объекты «бесплатно», как в Java или C++ — изоляция намеренная, она и убирает большинство гонок. Создание потока не бесплатно: поднимается новый изолят V8, на это уходят десятки миллисекунд и память (каждый worker — это мегабайты под собственный движок). Отсюда практическое правило: не создавайте по worker'у на каждый мелкий запрос. Делают пул воркеров — заранее поднятые потоки, между которыми раздаются задания (готовые реализации есть в пакете piscina). Поток живёт, пока есть незавершённая работа или открытый MessagePort; чтобы он не держал процесс, лишний канал закрывают или вызывают worker.terminate().

Частые ошибки

• Выносить в worker операции ввода-вывода. Чтение файла или запрос к БД и так асинхронны и не блокируют цикл — worker тут только добавит накладные расходы. Потоки нужны для CPU, не для I/O.
• Создавать новый Worker на каждый запрос. Старт потока — десятки миллисекунд и мегабайты памяти; под нагрузкой это убьёт сервер. Нужен пул.
• Ждать, что потоки делят переменные. У каждого своя куча; данные либо копируются через postMessage, либо живут в SharedArrayBuffer. Глобальную переменную из главного потока worker не увидит.
• Передавать буфер и продолжать им пользоваться. После transfer ([buf.buffer]) исходный буфер обнуляется — обращение к нему вернёт пустоту.
• Забыть про error и exit. Необработанная ошибка внутри потока без слушателя worker.on('error') молча потеряется, а Promise зависнет навсегда.

Итоги

• worker_threads дают настоящие потоки внутри одного процесса — для тяжёлых вычислений, которые иначе заблокируют событийный цикл.
• У каждого потока свой V8 и своя память; общение — через postMessage/parentPort, данные копируются алгоритмом structured clone.
• Крупные буферы можно передавать во владение (transfer) без копирования — указав их в списке вторым аргументом postMessage.
• SharedArrayBuffer + Atomics дают по-настоящему общую память с атомарными операциями, но возвращают классические риски гонок.
• Старт потока дорог, поэтому на практике используют пул воркеров, а I/O в потоки не выносят.