Как работает Worker Threads в Node.js?
Что такое Worker Threads
Модуль worker_threads появился в Node.js 10 (стабильно с 12) и позволяет выполнять JavaScript-код в настоящих OS-потоках внутри одного процесса. Это решает главную проблему Node.js — однопоточность event loop, которая становится узким местом при CPU-intensive задачах (парсинг, шифрование, обработка изображений, ML-инференс).
Архитектура
Каждый воркер запускает собственный экземпляр V8 и event loop, но находится в том же процессе ОС. Это означает:
- Общая память через
SharedArrayBufferиAtomics - Передача данных через structured clone (копирование) или Transferable Objects (zero-copy)
- Изоляция: воркер не имеет доступа к глобальному
process, у него своя копия модульной системы
Способы передачи данных
Structured Clone (по умолчанию)
Данные сериализуются и копируются в память воркера. Безопасно, но при больших объёмах создаёт накладные расходы.
Transferable Objects
Объекты типа ArrayBuffer передаются без копирования — право владения переходит к принимающей стороне, исходный буфер становится непригодным к использованию.
SharedArrayBuffer
Оба потока работают с одной областью памяти. Синхронизация осуществляется через Atomics — атомарные операции чтения/записи и примитивы Atomics.wait() / Atomics.notify().
Когда использовать
- CPU-bound задачи: парсинг JSON большого объёма, сжатие, криптография
- Параллельная обработка независимых кусков данных
- Изоляция «тяжёлого» кода, чтобы не блокировать event loop основного потока
Не подходит для I/O-задач — там достаточно стандартной асинхронности Node.js.
Worker Pool pattern
Создавать воркер на каждый запрос дорого (~2–5 мс). На практике используют пул: workerpool, piscina или самодельный менеджер, который переиспользует уже запущенные потоки.
Ограничения
- Нет доступа к DOM (не актуально для Node.js, но важно при переносе кода из браузера)
require()/importнезависимы в каждом воркере — нет singleton-гарантий- Ошибка внутри воркера не падает в main thread автоматически, нужно слушать событие
error
Сравнение с альтернативами
| Механизм | Процесс | IPC | Память | Latency |
|---|---|---|---|---|
child_process / cluster | Новый | Pipe/Socket | Изолирована | Высокая |
worker_threads | Тот же | MessageChannel | Разделяемая | Низкая |
| libuv thread pool | Тот же | Внутренний | Нет JS-доступа | Минимальная |
Что хочет услышать интервьюер
Понимание разницы между worker_threads и child_process/cluster — один процесс vs новый процесс, разделяемая память vs изолированная
Знание трёх способов обмена данными: structured clone, Transferable Objects, SharedArrayBuffer с Atomics
Чёткое понимание, при каких задачах Worker Threads даёт выигрыш (CPU-bound), а при каких бесполезен (I/O)
Осознание накладных расходов на создание воркера и знание паттерна Worker Pool (piscina, workerpool)
Понимание, что каждый воркер имеет свой event loop и свою копию V8, что влечёт изоляцию модулей и отдельную обработку ошибок
Пример: Базовый пример: main thread и воркер
// main.ts
import { Worker, isMainThread, parentPort, workerData } from 'worker_threads';
import path from 'path';
if (isMainThread) {
// Запускаем воркер, передаём данные через workerData (structured clone)
const worker = new Worker(__filename, {
workerData: { numbers: [1, 2, 3, 4, 5] },
});
worker.on('message', (result: number) => {
console.log('Сумма из воркера:', result); // 15
});
worker.on('error', (err) => {
console.error('Ошибка в воркере:', err);
});
worker.on('exit', (code) => {
if (code !== 0) console.error(`Воркер завершился с кодом ${code}`);
});
} else {
// Код воркера: считаем сумму и отправляем результат
const { numbers } = workerData as { numbers: number[] };
const sum = numbers.reduce((acc, n) => acc + n, 0);
parentPort!.postMessage(sum);
}
Пример: Transferable Objects: zero-copy передача ArrayBuffer
import { Worker, isMainThread, parentPort } from 'worker_threads';
if (isMainThread) {
const worker = new Worker(__filename);
// Создаём большой буфер (100 MB)
const buffer = new ArrayBuffer(100 * 1024 * 1024);
// Передаём без копирования — после этого buffer в main thread недоступен
worker.postMessage({ buffer }, [buffer]);
console.log('Буфер передан, byteLength теперь:', buffer.byteLength); // 0
} else {
parentPort!.on('message', ({ buffer }: { buffer: ArrayBuffer }) => {
console.log('Воркер получил буфер, размер:', buffer.byteLength);
// Обрабатываем данные...
});
}
Пример: SharedArrayBuffer + Atomics для синхронизации
import { Worker, isMainThread, parentPort, workerData } from 'worker_threads';
if (isMainThread) {
// Разделяемый буфер: индекс 0 — флаг готовности (0 = не готово, 1 = готово)
const sharedBuffer = new SharedArrayBuffer(4);
const flag = new Int32Array(sharedBuffer);
const worker = new Worker(__filename, { workerData: { sharedBuffer } });
// Ждём сигнала от воркера (блокирующий вызов, использовать осторожно)
Atomics.wait(flag, 0, 0); // ждём, пока flag[0] != 0
console.log('Main thread: воркер завершил работу');
} else {
const { sharedBuffer } = workerData as { sharedBuffer: SharedArrayBuffer };
const flag = new Int32Array(sharedBuffer);
// Имитируем долгую CPU-задачу
let sum = 0;
for (let i = 0; i < 1e8; i++) sum += i;
// Устанавливаем флаг и будим main thread
Atomics.store(flag, 0, 1);
Atomics.notify(flag, 0, 1);
}
Пример: Простой Worker Pool
import { Worker } from 'worker_threads';
import os from 'os';
type Task = {
data: unknown;
resolve: (value: unknown) => void;
reject: (reason?: unknown) => void;
};
class WorkerPool {
private workers: Worker[] = [];
private queue: Task[] = [];
private idleWorkers: Worker[] = [];
constructor(private workerFile: string, size = os.cpus().length) {
for (let i = 0; i < size; i++) {
this.addWorker();
}
}
private addWorker() {
const worker = new Worker(this.workerFile);
worker.on('message', (result) => {
const task = (worker as any).__currentTask as Task;
task.resolve(result);
this.idleWorkers.push(worker);
this.drain();
});
worker.on('error', (err) => {
const task = (worker as any).__currentTask as Task;
task.reject(err);
// Пересоздаём воркер после ошибки
this.workers = this.workers.filter((w) => w !== worker);
this.addWorker();
});
this.workers.push(worker);
this.idleWorkers.push(worker);
}
private drain() {
if (this.queue.length > 0 && this.idleWorkers.length > 0) {
const task = this.queue.shift()!;
const worker = this.idleWorkers.shift()!;
(worker as any).__currentTask = task;
worker.postMessage(task.data);
}
}
run(data: unknown): Promise<unknown> {
return new Promise((resolve, reject) => {
this.queue.push({ data, resolve, reject });
this.drain();
});
}
async terminate() {
await Promise.all(this.workers.map((w) => w.terminate()));
}
}
Типичные ошибки
Использование worker_threads для I/O-задач вместо обычного async/await — это не даёт прироста, но усложняет код
Создание нового воркера на каждый запрос без пула, что сводит на нет выигрыш по производительности из-за overhead на инициализацию
Игнорирование события 'error' на воркере — необработанная ошибка в воркере не крашит main thread, задача просто зависает
Попытка разделить обычные объекты через SharedArrayBuffer без Atomics, что приводит к race condition
Путаница с singleton-модулями: кеш require() не разделяется между воркерами, поэтому БД-соединение или глобальный стейт нужно создавать в каждом воркере отдельно
AI-тренажеры
4.7
Практика в студии
