Node.js традиційно проектувався як середовище з однопотоковим циклом подій, оптимізоване під операції введення-виведення (I/O-bound). Проте сучасні бекенд-застосунки дедалі частіше стикаються з обчислювально важкими завданнями (CPU-bound): валідація та трансформація великих структур JSON/Protocol Buffers, генерація PDF, компресія та шифрування медіа, розрахунок криптографічних гешів або інференс легких моделей машинного навчання. Виконання таких операцій у головному потоці блокує цикл подій libuv і призводить до вибухового зростання затримок (latency) для всіх підключених клієнтів.
Історично для паралелізму в 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) |
+-----------------------------------+-------------------------------------+
| |
+------------------+------------------+
|
+-------------------------------------------------+
| Спільна віртуальна пам'ять |
| - SharedArrayBuffer (Zero-copy concurrent R/W) |
| - Atomics Synchronization API (wait / notify) |
+-------------------------------------------------+Головна архітектурна відмінність Worker Threads від потоків у Java, C++ чи Go полягає в суворій ізоляції стану:
transferList) або спільної пам'яті SharedArrayBuffer із потокобезпечними атомарними операціями Atomics.Взаємодія між головним потоком та воркерами організована через канали передачі повідомлень (MessageChannel / MessagePort), які є абстракцією над асинхронними чергами завдань libuv:
[Головний потік: V8 Isolate 1] [Робочий потік: V8 Isolate 2]
| |
| --- postMessage({ task: 1 }) [Structured Clone] -> | (Десеріалізація в Heap 2)
| |
| --- postMessage(arrayBuffer, [arrayBuffer]) -----> | (Zero-copy: Detach Heap 1)
| (Передача володіння) |
| |
| <======== SharedArrayBuffer (Atomics) =========> | (Прямий паралельний доступ)port.postMessage(data) рушій V8 серіалізує граф об'єктів у внутрішній бінарний формат, передає байти в чергу цільового потоку та десеріалізує їх у купі приймача. Для масивних об'єктів (наприклад, 100 МБ) це витрачає відчутні ресурси CPU на копіювання.ArrayBuffer через список transferList:
port.postMessage({ buffer }, [buffer]);buffer.byteLength стає рівною 0, унеможливлюючи неузгоджені зміни.Atomics гарантують синхронізацію без стану гонитви.HTTP-контролер передає ресурсомістке завдання диспетчеру ThreadPool, отримуючи відкладений Promise.
Якщо в пулі є вільний воркер, завдання передається негайно. Якщо всі зайняті, воно очікує в черзі FIFO.
Бінарні структури передаються у воркер через transferList із миттєвим нульовим копіюванням (zero-copy).
Воркер виконує математичні операції на виділеному ядрі процесора, не блокуючи головний цикл подій.
Воркер надсилає результат головному потоку. Диспетчер резолвить проміс і бере наступне завдання з черги.
Для забезпечення безпечного доступу до спільної пам'яті будуються класичні примітиви синхронізації. За допомогою Atomics.wait() та Atomics.notify() можна реалізувати справжній м'ютекс взаємного виключення:
/**
* SharedMutex: М'ютекс на базі 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 {
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;
}
}Специфікація стандарту ECMAScript категорично забороняє виклик Atomics.wait() у головному потоці виконання Node.js. Спроба виконати цей метод кине синхронне виключення TypeError: Atomics.wait cannot be called in this context. Блокування головного потоку зупинило б цикл подій libuv, паралізуючи мережеві з'єднання, таймери та обробку запитів.
Використовуйте Atomics.wait виключно у фонових воркерах. У головному потоці для асинхронного очікування застосовуйте метод Atomics.waitAsync(), який повертає Promise.
Створення екземпляра new Worker('./worker.js') на кожен веб-запит спричиняє критичну деградацію сервісу. Кожен потік виділяє новий V8 Isolate, компілює скрипт заново та запускає окремий екземпляр циклу подій. Під навантаженням у 100 RPS сервер вичерпає ліміти оперативної пам'яті й буде вбитий операційною системою через OOM Killer.
Завжди використовуйте довгоживучий пул потоків (наприклад, бібліотеку 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() для гарантованої атомарності.
Головний потік створює довгоживучі воркери. ОС виділяє системні потоки, а Node.js ініціалізує незалежні рушії V8 та цикли libuv.
Диспетчер формує метадані та передає бінарні буфери з миттєвим відчуженням володіння (zero-copy) через transferList.
Воркер отримує подію message та виконує ресурсомісткі розрахунки на виділеному ядрі процесора без затримок для головного Event Loop.
Результат повертається головному потоку. Коллбек фази Poll завершує клієнтський Promise без зайвого блокування.
Воркер не знищується, а залишається активним у пулі, миттєво готовий прийняти наступне завдання з черги.