Пошук уроків, статей та іншого контенту
Розглянемо Worker Threads, обмін повідомленнями та випадки, коли винесення CPU-операцій в окремий потік виправдане.
Worker Threads — це механізм Node.js для виконання JavaScript-коду в окремих потоках операційної системи.
Звичайно Node.js виконує JavaScript в одному основному потоці. Його event loop добре обробляє асинхронні операції, але довгий синхронний CPU-розрахунок блокує весь процес:
нові HTTP-запити не обробляються;
таймери спрацьовують із затримкою;
обробники подій не виконуються;
зростає час відповіді застосунку.
Worker має власний:
event loop;
JavaScript-контекст;
стек викликів;
набір глобальних змінних;
об'єкт process.
За замовчуванням worker і основний потік не мають спільної JavaScript-пам'яті. Обмін виконується повідомленнями.
Основний потік ← повідомлення → WorkerЦе відрізняє worker threads від простого виклику функції в тому самому потоці.
Worker Threads корисні для операцій, які значний час використовують CPU:
складні математичні розрахунки;
обробка великих масивів даних;
кодування або декодування;
обчислення хешів;
аналіз тексту;
обробка зображень або аудіо;
синхронні операції над великими обсягами даних, якщо їх неможливо розбити на короткі частини.
Головний критерій — операція має бути CPU-bound, тобто обмеженою продуктивністю процесора.
Worker зазвичай не потрібен для:
HTTP-запитів;
запитів до бази даних;
читання файлів через асинхронний API;
мережевих операцій;
очікування таймера.
Такі операції вже можуть виконуватися асинхронно, не блокуючи event loop основного потоку.
Worker Thread не робить повільну операцію швидшою сам по собі. Він дозволяє виконати її паралельно з основним потоком і не заблокувати обробку інших подій.
node:worker_threadsДля роботи з потоками використовується вбудований модуль node:worker_threads.
Основні імпорти:
const {
Worker,
isMainThread,
parentPort,
workerData,
} = require('node:worker_threads');WorkerКлас Worker використовується для створення нового потоку:
const worker = new Worker('./worker.js');Worker може виконувати окремий файл або поточний файл:
const worker = new Worker(__filename);isMainThreadisMainThread має значення:
true у головному потоці;
false усередині worker.
Це дозволяє використовувати один файл і розділити логіку головного потоку та worker:
if (isMainThread) {
// Код головного потоку
} else {
// Код worker
}parentPortparentPort — це порт для обміну повідомленнями між worker і потоком, який його створив.
У worker:
parentPort.postMessage({ type: 'result', value: 42 });
parentPort.on('message', (message) => {
console.log(message);
});У головному потоці повідомлення доступні через події екземпляра Worker:
worker.postMessage({ type: 'calculate', value: 42 });
worker.on('message', (message) => {
console.log(message);
});workerDataworkerData містить дані, передані під час створення worker:
const worker = new Worker('./worker.js', {
workerData: {
name: 'calculation-worker',
},
});У worker:
console.log(workerData.name);workerData підходить для початкової конфігурації. Для подальшого обміну даними використовується postMessage.
Під час виклику postMessage Node.js серіалізує значення та передає його іншому контексту. Використовується алгоритм структурованого клонування.
Зазвичай можна передавати:
примітиви;
масиви;
звичайні об'єкти;
Map;
Set;
typed arrays;
ArrayBuffer та інші підтримувані типи.
Функції, екземпляри класів із важливим внутрішнім станом і деякі спеціальні об'єкти не передаються так, як звичайні дані.
Краще використовувати повідомлення з явним типом:
{
type: 'calculate',
requestId: 'request-123',
value: 42
}Це спрощує маршрутизацію кількох операцій і зіставлення відповіді із запитом.
Нижче worker отримує завдання для обчислення числа Фібоначчі. Розрахунок навмисно виконується рекурсивно та синхронно, щоб продемонструвати CPU-bound операцію.
Збережіть код у файл worker-demo.js і запустіть:
node worker-demo.js'use strict';
const {
Worker,
isMainThread,
parentPort,
workerData,
} = require('node:worker_threads');
function fibonacci(number) {
if (number <= 1) {
return number;
}
return fibonacci(number - 1) + fibonacci(number - 2);
}
if (isMainThread) {
const worker = new Worker(__filename, {
workerData: {
name: 'fibonacci-worker',
},
});
const timer = setInterval(() => {
console.log('Головний потік продовжує обробляти події');
}, 100);
worker.on('message', (message) => {
if (message.type === 'ready') {
console.log(`Worker "${message.name}" готовий`);
worker.postMessage({
type: 'calculate',
requestId: 'fib-1',
value: 42,
});
return;
}
if (message.type === 'result') {
console.log(
`Результат запиту ${message.requestId}: fibonacci(${message.value}) = ${message.result}`,
);
clearInterval(timer);
worker.terminate();
}
if (message.type === 'error') {
console.error(`Помилка розрахунку: ${message.message}`);
clearInterval(timer);
worker.terminate();
}
});
worker.on('error', (error) => {
console.error('Непередбачена помилка worker:', error);
clearInterval(timer);
});
worker.on('exit', (code) => {
if (code !== 0) {
console.error(`Worker завершився з кодом ${code}`);
} else {
console.log('Worker завершив роботу');
}
});
} else {
parentPort.postMessage({
type: 'ready',
name: workerData.name,
});
parentPort.on('message', (message) => {
if (message.type !== 'calculate') {
return;
}
try {
const result = fibonacci(message.value);
parentPort.postMessage({
type: 'result',
requestId: message.requestId,
value: message.value,
result,
});
} catch (error) {
parentPort.postMessage({
type: 'error',
requestId: message.requestId,
message: error.message,
});
}
});
}У цьому прикладі:
Головний потік створює worker.
workerData передає worker його ім'я.
Worker надсилає повідомлення ready.
Головний потік надсилає завдання через worker.postMessage.
Worker виконує синхронний розрахунок.
Worker повертає результат через parentPort.postMessage.
Головний потік продовжує працювати під час розрахунку.
Після завершення worker зупиняється через worker.terminate().
Повідомлення з type, requestId та іншими полями утворюють простий протокол між двома потоками.
Після створення worker можна слухати кілька важливих подій.
messageСпрацьовує, коли worker надсилає повідомлення:
worker.on('message', (message) => {
console.log(message);
});errorСпрацьовує, якщо всередині worker виникла необроблена помилка:
worker.on('error', (error) => {
console.error(error);
});Ця подія важлива для логування та обробки відмови worker.
exitСпрацьовує після завершення worker:
worker.on('exit', (code) => {
console.log(`Worker завершився з кодом ${code}`);
});Код 0 зазвичай означає нормальне завершення. Ненульовий код може вказувати на помилку або аварійне завершення.
terminate()Метод terminate() примусово завершує worker:
await worker.terminate();Метод повертає Promise. Його варто використовувати для очищення worker, який більше не потрібен.
Якщо worker має обробляти багато завдань, не слід створювати новий потік для кожного невеликого повідомлення. Краще залишити worker активним і надсилати йому кілька завдань.
Події зручні для постійного обміну повідомленнями. Для одноразового завдання можна створити Promise-обгортку:
const { Worker } = require('node:worker_threads');
function runWorker(value) {
return new Promise((resolve, reject) => {
const worker = new Worker('./calculate-worker.js', {
workerData: { value },
});
worker.once('message', resolve);
worker.once('error', reject);
worker.once('exit', (code) => {
if (code !== 0) {
reject(new Error(`Worker завершився з кодом ${code}`));
}
});
});
}Таку функцію можна використовувати через async/await, але потрібно враховувати вартість створення потоку.
Worker — це не безкоштовний виклик функції. Під час його створення Node.js має:
ініціалізувати новий JavaScript-контекст;
запустити event loop;
завантажити необхідний код;
виділити пам'ять;
налаштувати канали обміну.
Тому створення нового worker для кожного маленького завдання може бути повільнішим, ніж виконання цього завдання в основному потоці.
Для великої кількості CPU-завдань зазвичай використовують пул worker-потоків:
Створюють обмежену кількість worker.
Додають завдання до черги.
Вільний worker отримує наступне завдання.
Після завершення worker повертається до пулу.
Кількість worker обмежують відповідно до кількості доступних ядер і характеру навантаження.
Пул запобігає створенню сотень потоків і неконтрольованому споживанню пам'яті.
Звичайні повідомлення копіюються або клонуються. Для великих бінарних даних це може бути дорого.
ArrayBuffer можна передати через список transfer-об'єктів:
worker.postMessage(
{
type: 'process-buffer',
buffer,
},
[buffer],
);Після передавання власність над буфером переходить до worker. У головному потоці такий ArrayBuffer стає від'єднаним і більше не може використовуватися як раніше.
Це допомагає уникати копіювання, але вимагає уважно контролювати життєвий цикл буфера.
Для спільного доступу до пам'яті існують SharedArrayBuffer та Atomics, однак вони додають складність синхронізації. Їх варто використовувати лише тоді, коли копіювання або передавання даних справді стало вузьким місцем.
Worker Threads та окремі процеси розв'язують різні задачі:
worker працює в тому самому процесі Node.js, але в окремому потоці;
окремий процес має власну пам'ять і межу відмови;
worker зручний для CPU-операцій із передаванням повідомлень;
процеси краще ізолюють ресурси та помилки.
У межах цієї теми важливо запам'ятати: Worker Threads — це спосіб винести обчислення з event loop, а не універсальна заміна процесам чи асинхронному I/O.
Передавання великих об'єктів може вимагати значного часу та пам'яті через клонування. Для великих бінарних даних варто розглянути transfer-об'єкти.
Не всі значення можна передати через postMessage. Повідомлення мають містити дані, а не поведінку. Функцію не можна передати worker і виконати там як звичайне поле об'єкта.
Помилка worker не повинна залишати основний потік у стані нескінченного очікування. Код, який очікує результат, має передбачати:
повідомлення про помилку;
подію error;
аварійне завершення worker;
можливість повторити або відхилити завдання.
Довгоживучі worker потрібно коректно завершувати під час зупинки застосунку. Інакше вони можуть утримувати ресурси процесу.
Створення потоку для кожного HTTP-запиту створює зайві накладні витрати.
Краще:
використовувати пул worker;
групувати завдання;
не передавати в worker операції, які виконуються дуже швидко.
Worker не потрібен лише тому, що операція пов'язана з файлом або мережею. Асинхронний API Node.js уже дозволяє не блокувати event loop.
Потреба у worker виникає тоді, коли саме JavaScript-обчислення надовго займає процесор.
Worker також має власний event loop. Якщо в ньому послідовно виконувати кілька довгих синхронних операцій, повідомлення до нього можуть чекати в черзі.
Для паралельних незалежних обчислень потрібні кілька worker або пул.
Якщо основний потік надсилає завдання швидше, ніж worker їх обробляє, черга повідомлень зростатиме.
Потрібно контролювати:
кількість завдань у черзі;
максимальну кількість одночасних обчислень;
час очікування;
скасування або відхилення надлишкових завдань.
Глобальні змінні основного потоку недоступні worker. Так само зміни глобальної змінної worker не з'являються в основному потоці.
Дані потрібно передавати через:
workerData;
postMessage;
спеціальні механізми спільної пам'яті.
Якщо одночасно виконуються кілька завдань, відповідь без requestId може бути складно зіставити з початковим запитом.
Для протоколу обміну використовуйте унікальний ідентифікатор завдання.
Worker Threads виконують JavaScript у окремих потоках.
Вони призначені насамперед для CPU-bound операцій.
Основний потік і worker обмінюються повідомленнями через postMessage.
У worker використовується parentPort, а в головному потоці — екземпляр Worker.
workerData передає початкові параметри під час створення потоку.
Під час обміну дані зазвичай клонуються, а великі ArrayBuffer можна передавати без копіювання.
Для багатьох завдань краще використовувати пул worker, а не створювати потік для кожної операції.
Асинхронний I/O зазвичай не потребує Worker Threads.
Надійна реалізація має обробляти повідомлення, помилки, аварійне завершення та очищення worker.