Node.js традиционно проектировался как платформа с однопоточным циклом событий, оптимизированная под ввод-вывод (I/O-bound). Однако современные серверные приложения все чаще сталкиваются с вычислительно тяжелыми задачами (CPU-bound): валидация и трансформация больших JSON/Protocol Buffers структур, генерация PDF, сжатие и шифрование данных, расчет криптографических хэшей или инференс легковесных ML-моделей. Выполнение таких операций в главном потоке приводит к блокировке цикла событий libuv и взрывному росту латентности.
Исторически для параллелизма в Node.js применялись процессы операционной системы: модули child_process (fork, spawn) и cluster. Начиная с Node.js 10.5+ и стабилизации в Node.js 12+, платформа предоставляет модуль worker_threads, реализующий настоящую многопоточность операционной системы внутри единого процесса:
+-------------------------------------------------------------------------+
| Single OS Process (PID: 14205) |
+-------------------------------------------------------------------------+
| Main Thread | Worker Thread #1 |
| - libuv Event Loop | - libuv Event Loop |
| - V8 Isolate A (Heap A, Stack A) | - V8 Isolate B (Heap B, Stack B) |
| - Call Stack | - Call Stack |
+-----------------------------------+-------------------------------------+
| |
+------------------+------------------+
|
+-------------------------------------------------+
| Shared Virtual Memory |
| - SharedArrayBuffer (Zero-copy concurrent R/W) |
| - Atomics Synchronization API (wait / notify) |
+-------------------------------------------------+Главное архитектурное отличие Worker Threads от потоков в Java, C++ или Go заключается в изоляции состояния:
transferList), либо через разделяемую память SharedArrayBuffer с потокобезопасными операциями Atomics.Взаимодействие главного потока и воркеров строится вокруг каналов передачи сообщений (MessageChannel / MessagePort), которые представляют собой абстракцию над асинхронными очередями libuv:
[Main Thread: V8 Isolate 1] [Worker Thread: V8 Isolate 2]
| |
| --- postMessage({ task: 1 }) [Structured Clone] -> | (Десериализация в Heap 2)
| |
| --- postMessage(arrayBuffer, [arrayBuffer]) -----> | (Zero-copy: Detach Heap 1)
| (Transfer ownership) |
| |
| <======== SharedArrayBuffer (Atomics) =========> | (Прямой параллельный доступ)Главный поток создает объект Worker. Операционная система аллоцирует поток ядра, а Node.js инициализирует отдельный V8 Isolate и libuv loop.
Диспетчер сериализует метаданные и передает бинарные входные структуры с отчуждением владения через transferList.
Воркер принимает событие message, производит ресурсоемкие математические вычисления на выделенном ядре процессора.
Результат отправляется обратно через postMessage. В главном потоке срабатывает коллбэк фазы Poll, завершающий Promise.
Воркер не уничтожается, а остается в пуле готовых потоков в ожидании следующего сообщения из очереди диспетчера.
Почему передача крупного ArrayBuffer через transferList во второй аргумент postMessage происходит с нулевыми накладными расходами (Zero-Copy)?
port.postMessage(data) движок V8 сериализует граф объектов во внутренний бинарный формат, передает массив байтов в очередь другого потока и десериализует его в куче принимающего потока. Для крупных объектов (например, 100 МБ) это отнимает значительное время CPU на сериализацию.ArrayBuffer через список передачи:
port.postMessage({ buffer }, [buffer]);buffer.byteLength становится равным 0, что исключает возможность одновременной несогласованной модификации.Atomics.HTTP-контроллер отправляет тяжелую задачу вычисления хэша в класс ThreadPool, возвращающий Promise.
Если в пуле есть свободный воркер, задача отдается ему немедленно. Если все заняты, задача встает в очередь FIFO.
Входные бинарные массивы передаются в воркер через transferList с нулевым копированием (zero-copy).
Поток выполняет алгоритм на 100% мощности выделенного ядра CPU, не потребляя ресурсы главного Event Loop.
Воркер отправляет результат обратно главному потоку. Диспетчер разрешает промис и немедленно берет следующую задачу из очереди.
В многопоточном коде без общей памяти невозможно построить быстрые примитивы синхронизации. С использованием Atomics.wait() и Atomics.notify() можно реализовать настоящий мьютекс (взаимное исключение):
/**
* Блокировка (Мьютекс) на базе SharedArrayBuffer и Atomics.
* Индекс 0: состояние замка (0 = UNLOCKED, 1 = LOCKED).
*/
export class SharedMutex {
private static readonly UNLOCKED = 0;
private static readonly LOCKED = 1;
constructor(private readonly sharedBuffer: Int32Array) {}
public static createBuffer(): Int32Array {
// 4 байта для Int32
const sab = new SharedArrayBuffer(4);
return new Int32Array(sab);
}
public lock(): void {
// Атомарно пытаемся установить состояние LOCKED, если текущее UNLOCKED
while (
Atomics.compareExchange(
this.sharedBuffer,
0,
SharedMutex.UNLOCKED,
SharedMutex.LOCKED
) !== SharedMutex.UNLOCKED
) {
// Если мьютекс уже захвачен, рабочий поток засыпает в ядре ОС
// ВНИМАНИЕ: Atomics.wait запрещен в главном потоке Node.js!
Atomics.wait(this.sharedBuffer, 0, SharedMutex.LOCKED);
}
}
public unlock(): void {
// Сбрасываем состояние в UNLOCKED
const oldValue = Atomics.exchange(this.sharedBuffer, 0, SharedMutex.UNLOCKED);
if (oldValue === SharedMutex.LOCKED) {
// Пробуждаем один поток, ожидающий на данном адресе
Atomics.notify(this.sharedBuffer, 0, 1);
}
}
public runSynchronized<T>(action: () => T): T {
this.lock();
try {
return action();
} finally {
this.unlock();
}
}
}Создание нового Worker на каждый запрос — грубейший антипаттерн, приводящий к колоссальному оверхеду. На практике развертывается пул потоков с фиксированным числом воркеров, равным количеству физических ядер сервера.
import { Worker } from 'node:worker_threads';
import os from 'node:os';
import path from 'node:path';
interface Task<T, R> {
id: string;
data: T;
resolve: (result: R) => void;
reject: (err: Error) => void;
timeoutTimer: NodeJS.Timeout;
}
interface WorkerRecord {
worker: Worker;
busy: boolean;
activeTaskId: string | null;
}
export class ProductionThreadPool<T = unknown, R = unknown> {
private readonly workers: WorkerRecord[] = [];
private readonly taskQueue: Task<T, R>[] = [];
private isTerminating = false;
constructor(
private readonly workerScriptPath: string,
private readonly poolSize: number = os.cpus().length,
private readonly taskTimeoutMs: number = 30_000
) {
this.initPool();
}
private initPool(): void {
for (let i = 0; i < this.poolSize; i++) {
this.spawnWorker();
}
}
private spawnWorker(): void {
const worker = new Worker(this.workerScriptPath);
const record: WorkerRecord = {
worker,
busy: false,
activeTaskId: null,
};
worker.on('message', (result: { taskId: string; payload: R; error?: string }) => {
this.handleWorkerResult(record, result);
});
worker.on('error', (err) => {
console.error(`[Worker Error] Поток аварийно завершился:`, err);
this.replaceCrashedWorker(record);
});
this.workers.push(record);
}
public execute(data: T): Promise<R> {
if (this.isTerminating) {
return Promise.reject(new Error('Пул потоков находится в состоянии завершения работы'));
}
return new Promise<R>((resolve, reject) => {
const taskId = `task_${Date.now()}_${Math.random().toString(36).slice(2, 9)}`;
const timeoutTimer = setTimeout(() => {
this.handleTaskTimeout(taskId);
}, this.taskTimeoutMs);
const task: Task<T, R> = {
id: taskId,
data,
resolve,
reject,
timeoutTimer,
};
this.taskQueue.push(task);
this.dispatchNext();
});
}
private dispatchNext(): void {
if (this.taskQueue.length === 0) return;
// Поиск первого свободного воркера
const availableWorker = this.workers.find((w) => !w.busy);
if (!availableWorker) return;
const task = this.taskQueue.shift()!;
availableWorker.busy = true;
availableWorker.activeTaskId = task.id;
// Сохраняем обработчик текущей задачи в замыкании воркера
(availableWorker.worker as any).__currentTask = task;
availableWorker.worker.postMessage({
taskId: task.id,
data: task.data,
});
}
private handleWorkerResult(
record: WorkerRecord,
result: { taskId: string; payload: R; error?: string }
): void {
const currentTask = (record.worker as any).__currentTask as Task<T, R> | undefined;
if (currentTask && currentTask.id === result.taskId) {
clearTimeout(currentTask.timeoutTimer);
if (result.error) {
currentTask.reject(new Error(result.error));
} else {
currentTask.resolve(result.payload);
}
(record.worker as any).__currentTask = undefined;
}
record.busy = false;
record.activeTaskId = null;
this.dispatchNext();
}
private handleTaskTimeout(taskId: string): void {
const workerIndex = this.workers.findIndex((w) => w.activeTaskId === taskId);
if (workerIndex === -1) return;
const record = this.workers[workerIndex];
const currentTask = (record.worker as any).__currentTask as Task<T, R>;
if (currentTask) {
currentTask.reject(new Error(`Превышен лимит времени выполнения задачи (${this.taskTimeoutMs} мс)`));
}
// Принудительно уничтожаем зависший воркер и пересоздаем его
record.worker.terminate().finally(() => {
this.workers.splice(workerIndex, 1);
this.spawnWorker();
this.dispatchNext();
});
}
private replaceCrashedWorker(deadRecord: WorkerRecord): void {
const idx = this.workers.indexOf(deadRecord);
if (idx !== -1) {
this.workers.splice(idx, 1);
}
const currentTask = (deadRecord.worker as any).__currentTask as Task<T, R> | undefined;
if (currentTask) {
clearTimeout(currentTask.timeoutTimer);
currentTask.reject(new Error('Рабочий поток аварийно завершился во время исполнения'));
}
if (!this.isTerminating) {
this.spawnWorker();
this.dispatchNext();
}
}
public async shutdown(): Promise<void> {
this.isTerminating = true;
// Отклоняем все задачи, оставшиеся в очереди
while (this.taskQueue.length > 0) {
const task = this.taskQueue.shift()!;
clearTimeout(task.timeoutTimer);
task.reject(new Error('Пул потоков остановлен до начала исполнения задачи'));
}
// Ожидаем завершения всех потоков
const terminations = this.workers.map((w) => w.worker.terminate());
await Promise.all(terminations);
this.workers.length = 0;
}
}Почему синхронный вызов Atomics.wait категорически запрещен в главном потоке Node.js приложения?
Спецификация стандарта ECMAScript строго запрещает вызов Atomics.wait() в главном потоке Node.js (так же, как и в главном потоке браузера). Попытка выполнить этот метод выбросит синхронное исключение TypeError: Atomics.wait cannot be called in this context. Блокировка главного потока на примитиве синхронизации парализовала бы цикл событий libuv, включая обработку сетевых соединений.
Используйте Atomics.wait исключительно внутри фоновых потоков Worker. В главном потоке для асинхронного ожидания доступен метод Atomics.waitAsync() (возвращает Promise).
Создание объекта new Worker('./worker.js') на каждый веб-запрос приводит к тяжелой деградации сервиса. Каждый поток выделяет новый V8 Isolate, компилирует JavaScript-скрипт с нуля и запускает собственный инстанс libuv. При нагрузке в 100 RPS сервер мгновенно исчерпает лимиты оперативной памяти и обрушит процесс.
Всегда используйте паттерн долгоживущего пула потоков (Thread Pool, например библиотеку piscina) с количеством воркеров, ограниченным доступными физическими ядрами CPU (os.cpus().length).
Передача крупных объектов (буферы изображений, дампы данных более 50 МБ) через worker.postMessage(data) без передачи владения заставляет V8 полностью клонировать память через Structured Clone Algorithm. Это удваивает потребление оперативной памяти и тратит сотни миллисекунд процессорного времени на сериализацию.
Всегда передавайте бинарные данные в виде ArrayBuffer через второй аргумент transferList: port.postMessage(buffer, [buffer.buffer]).
Прямая запись несколькими потоками в один и тот же индекс SharedArrayBuffer (shared[0]++ или shared[0] = value) не является атомарной. На уровне процессора операция инкремента состоит из трех микроинструкций: чтение из памяти в регистр, увеличение регистра и запись обратно в память. При параллельном выполнении значение будет перезаписано со сбоем подсчета.
Всегда используйте функции Atomics.add(), Atomics.store(), Atomics.compareExchange() для гарантированной потокобезопасности.