Потоки Worker

Модуль node:worker_threads позволяет использовать потоки, в которых JavaScript выполняется параллельно. Чтобы получить к нему доступ:

import worker_threads from 'node:worker_threads';

const worker_threads = require('node:worker_threads');

Рабочие потоки (потоки) полезны для выполнения ресурсоёмких операций JavaScript, интенсивно использующих ЦП. Они мало помогают при выполнении операций, интенсивно использующих ввод-вывод. Встроенные асинхронные операции ввода-вывода Node.js эффективнее, чем Worker-потоки.

В отличие от child_process или cluster, worker_threads могут совместно использовать память. Это достигается передачей экземпляров ArrayBuffer или совместным использованием экземпляров SharedArrayBuffer.

import {
  Worker,
  isMainThread,
  parentPort,
  workerData,
} from 'node:worker_threads';

if (!isMainThread) {
  const { parse } = await import('some-js-parsing-library');
  const script = workerData;
  parentPort.postMessage(parse(script));
}

export default function parseJSAsync(script) {
  return new Promise((resolve, reject) => {
    const worker = new Worker(new URL(import.meta.url), {
      workerData: script,
    });
    worker.on('message', resolve);
    worker.once('error', reject);
    worker.once('exit', (code) => {
      if (code !== 0)
        reject(new Error(`Worker stopped with exit code ${code}`));
    });
  });
};

const {
  Worker,
  isMainThread,
  parentPort,
  workerData,
} = require('node:worker_threads');

if (isMainThread) {
  module.exports = function parseJSAsync(script) {
    return new Promise((resolve, reject) => {
      const worker = new Worker(__filename, {
        workerData: script,
      });
      worker.on('message', resolve);
      worker.once('error', reject);
      worker.once('exit', (code) => {
        if (code !== 0)
          reject(new Error(`Worker stopped with exit code ${code}`));
      });
    });
  };
} else {
  const { parse } = require('some-js-parsing-library');
  const script = workerData;
  parentPort.postMessage(parse(script));
}

В приведённом выше примере для каждого вызова parseJSAsync() создаётся поток Worker. На практике для таких задач следует использовать пул Worker-потоков. В противном случае затраты на создание Worker-потоков, скорее всего, превысят получаемую от них пользу.

При реализации пула Worker-потоков используйте API AsyncResource, чтобы сообщать диагностическим инструментам (например, для построения асинхронных трассировок стека) о связи между задачами и их результатами. Пример реализации см. в разделе «Использование AsyncResource для пула потоков Worker» документации async_hooks.

По умолчанию Worker-потоки наследуют параметры, не относящиеся к конкретному процессу. См. Worker constructor options, чтобы узнать, как настраивать параметры Worker-потока, в частности параметры argv и execArgv.

worker_threads.getEnvironmentData(key)

• key <any> Любое произвольное клонируемое значение JavaScript, которое можно использовать в качестве ключа <Map>.
• Возвращает: <any>

В Worker-потоке worker.getEnvironmentData() возвращает копию данных, переданных в worker.setEnvironmentData() потока, создавшего этот поток. Каждый новый Worker автоматически получает собственную копию данных окружения.

import {
  Worker,
  isMainThread,
  setEnvironmentData,
  getEnvironmentData,
} from 'node:worker_threads';

if (isMainThread) {
  setEnvironmentData('Hello', 'World!');
  const worker = new Worker(new URL(import.meta.url));
} else {
  console.log(getEnvironmentData('Hello'));  // Prints 'World!'.
}

const {
  Worker,
  isMainThread,
  setEnvironmentData,
  getEnvironmentData,
} = require('node:worker_threads');

if (isMainThread) {
  setEnvironmentData('Hello', 'World!');
  const worker = new Worker(__filename);
} else {
  console.log(getEnvironmentData('Hello'));  // Prints 'World!'.
}

worker_threads.isInternalThread

• Тип: <boolean>

Равно true, если этот код выполняется во внутреннем потоке Worker (например, в потоке загрузчика).

// loader.js
import { isInternalThread } from 'node:worker_threads';
console.log(isInternalThread);  // true

// loader.js
const { isInternalThread } = require('node:worker_threads');
console.log(isInternalThread);  // true

// main.js
import { isInternalThread } from 'node:worker_threads';
console.log(isInternalThread);  // false

// main.js
const { isInternalThread } = require('node:worker_threads');
console.log(isInternalThread);  // false

worker_threads.isMainThread

• Тип: <boolean>

Равно true, если этот код выполняется не в потоке Worker.

import { Worker, isMainThread } from 'node:worker_threads';

if (isMainThread) {
  // This re-loads the current file inside a Worker instance.
  new Worker(new URL(import.meta.url));
} else {
  console.log('Inside Worker!');
  console.log(isMainThread);  // Prints 'false'.
}

const { Worker, isMainThread } = require('node:worker_threads');

if (isMainThread) {
  // This re-loads the current file inside a Worker instance.
  new Worker(__filename);
} else {
  console.log('Inside Worker!');
  console.log(isMainThread);  // Prints 'false'.
}

worker_threads.markAsUntransferable(object)

• object <any> Любое произвольное значение JavaScript.

Помечает объект как непередаваемый. Если object присутствует в списке передачи вызова port.postMessage(), будет выброшена ошибка. Ничего не делает, если object — примитивное значение.

Это особенно полезно для объектов, которые можно клонировать, а не передавать, и которые используются другими объектами на стороне отправителя. Например, Node.js помечает таким образом ArrayBuffer, используемые для пула Buffer. Для таких экземпляров буфера массива вызов ArrayBuffer.prototype.transfer() запрещён.

Эту операцию нельзя отменить.

import { MessageChannel, markAsUntransferable } from 'node:worker_threads';

const pooledBuffer = new ArrayBuffer(8);
const typedArray1 = new Uint8Array(pooledBuffer);
const typedArray2 = new Float64Array(pooledBuffer);

markAsUntransferable(pooledBuffer);

const { port1 } = new MessageChannel();
try {
  // This will throw an error, because pooledBuffer is not transferable.
  port1.postMessage(typedArray1, [ typedArray1.buffer ]);
} catch (error) {
  // error.name === 'DataCloneError'
}

// The following line prints the contents of typedArray1 -- it still owns
// its memory and has not been transferred. Without
// `markAsUntransferable()`, this would print an empty Uint8Array and the
// postMessage call would have succeeded.
// typedArray2 is intact as well.
console.log(typedArray1);
console.log(typedArray2);

const { MessageChannel, markAsUntransferable } = require('node:worker_threads');

const pooledBuffer = new ArrayBuffer(8);
const typedArray1 = new Uint8Array(pooledBuffer);
const typedArray2 = new Float64Array(pooledBuffer);

markAsUntransferable(pooledBuffer);

const { port1 } = new MessageChannel();
try {
  // This will throw an error, because pooledBuffer is not transferable.
  port1.postMessage(typedArray1, [ typedArray1.buffer ]);
} catch (error) {
  // error.name === 'DataCloneError'
}

// The following line prints the contents of typedArray1 -- it still owns
// its memory and has not been transferred. Without
// `markAsUntransferable()`, this would print an empty Uint8Array and the
// postMessage call would have succeeded.
// typedArray2 is intact as well.
console.log(typedArray1);
console.log(typedArray2);

В браузерах нет аналога этого API.

worker_threads.isMarkedAsUntransferable(object)

• object <any> Любое значение JavaScript.
• Возвращает: <boolean>

Проверяет, помечен ли объект как непередаваемый с помощью markAsUntransferable().

import { markAsUntransferable, isMarkedAsUntransferable } from 'node:worker_threads';

const pooledBuffer = new ArrayBuffer(8);
markAsUntransferable(pooledBuffer);

isMarkedAsUntransferable(pooledBuffer);  // Returns true.

const { markAsUntransferable, isMarkedAsUntransferable } = require('node:worker_threads');

const pooledBuffer = new ArrayBuffer(8);
markAsUntransferable(pooledBuffer);

isMarkedAsUntransferable(pooledBuffer);  // Returns true.

В браузерах нет аналога этого API.

worker_threads.markAsUncloneable(object)

• object <any> Любое произвольное значение JavaScript.

Помечает объект как неклонируемый. Если object используется в качестве message в вызове port.postMessage(), будет выброшена ошибка. Ничего не делает, если object — примитивное значение.

Это не влияет на ArrayBuffer и любые объекты типа Buffer.

Эту операцию нельзя отменить.

import { markAsUncloneable } from 'node:worker_threads';

const anyObject = { foo: 'bar' };
markAsUncloneable(anyObject);
const { port1 } = new MessageChannel();
try {
  // This will throw an error, because anyObject is not cloneable.
  port1.postMessage(anyObject);
} catch (error) {
  // error.name === 'DataCloneError'
}

const { markAsUncloneable } = require('node:worker_threads');

const anyObject = { foo: 'bar' };
markAsUncloneable(anyObject);
const { port1 } = new MessageChannel();
try {
  // This will throw an error, because anyObject is not cloneable.
  port1.postMessage(anyObject);
} catch (error) {
  // error.name === 'DataCloneError'
}

В браузерах нет аналога этого API.

worker_threads.moveMessagePortToContext(port, contextifiedSandbox)

• port <MessagePort> Порт сообщений для передачи.

• contextifiedSandbox <Object> Контекстуализированный объект, возвращённый методом vm.createContext().

• Возвращает: <MessagePort>

Передаёт MessagePort в другой контекст vm. Исходный объект port становится непригодным для использования, а его место занимает возвращённый экземпляр MessagePort.

Возвращённый MessagePort является объектом в целевом контексте и наследуется от глобального класса Object этого контекста. Объекты, переданные обработчику события port.onmessage(), также создаются в целевом контексте и наследуются от его глобального класса Object.

Однако созданный MessagePort больше не наследуется от <EventTarget>, и для получения событий с его помощью можно использовать только port.onmessage().

worker_threads.parentPort

• Тип: <null> | <MessagePort>

Если этот поток является потоком Worker, это MessagePort, позволяющий обмениваться данными с родительским потоком. Сообщения, отправленные с помощью parentPort.postMessage(), доступны в родительском потоке через worker.on('message'), а сообщения, отправленные из родительского потока с помощью worker.postMessage(), доступны в этом потоке через parentPort.on('message').

import { Worker, isMainThread, parentPort } from 'node:worker_threads';

if (isMainThread) {
  const worker = new Worker(new URL(import.meta.url));
  worker.once('message', (message) => {
    console.log(message);  // Prints 'Hello, world!'.
  });
  worker.postMessage('Hello, world!');
} else {
  // When a message from the parent thread is received, send it back:
  parentPort.once('message', (message) => {
    parentPort.postMessage(message);
  });
}

const { Worker, isMainThread, parentPort } = require('node:worker_threads');

if (isMainThread) {
  const worker = new Worker(__filename);
  worker.once('message', (message) => {
    console.log(message);  // Prints 'Hello, world!'.
  });
  worker.postMessage('Hello, world!');
} else {
  // When a message from the parent thread is received, send it back:
  parentPort.once('message', (message) => {
    parentPort.postMessage(message);
  });
}

worker_threads.postMessageToThread(threadId, value[, transferList][, timeout])

• threadId <number> Идентификатор целевого потока. Если идентификатор потока недействителен, будет выброшена ошибка ERR_WORKER_MESSAGING_FAILED. Если идентификатор целевого потока совпадает с идентификатором текущего потока, будет выброшена ошибка ERR_WORKER_MESSAGING_SAME_THREAD.
• value <any> Отправляемое значение.
• transferList <Object[]> Если в value передан один или несколько объектов типа MessagePort, для этих элементов требуется transferList, иначе будет выброшена ошибка ERR_MISSING_MESSAGE_PORT_IN_TRANSFER_LIST. Дополнительные сведения см. в разделе port.postMessage().
• timeout <number> Время ожидания доставки сообщения в миллисекундах. По умолчанию — undefined, то есть ожидание продолжается бесконечно. Если время ожидания операции истечёт, будет выброшена ошибка ERR_WORKER_MESSAGING_TIMEOUT.
• Возвращает: <Promise> Промис, который выполняется, если сообщение успешно обработано целевым потоком.

Отправляет значение другому Worker-потоку, определяемому по его идентификатору.

Если в целевом потоке нет обработчика события workerMessage, операция выбросит ошибку ERR_WORKER_MESSAGING_FAILED.

Если в целевом потоке при обработке события workerMessage возникла ошибка, операция выбросит ошибку ERR_WORKER_MESSAGING_ERRORED.

Этот метод следует использовать, если целевой поток не является непосредственным родительским или дочерним потоком текущего. Если потоки находятся в отношении родитель-потомок, используйте require('node:worker_threads').parentPort.postMessage() и worker.postMessage(), чтобы обеспечить обмен данными между потоками.

В приведённом ниже примере показано использование postMessageToThread: создаются 10 вложенных потоков, и последний пытается связаться с главным потоком.

import process from 'node:process';
import {
  postMessageToThread,
  threadId,
  workerData,
  Worker,
} from 'node:worker_threads';

const channel = new BroadcastChannel('sync');
const level = workerData?.level ?? 0;

if (level < 10) {
  const worker = new Worker(new URL(import.meta.url), {
    workerData: { level: level + 1 },
  });
}

if (level === 0) {
  process.on('workerMessage', (value, source) => {
    console.log(`${source} -> ${threadId}:`, value);
    postMessageToThread(source, { message: 'pong' });
  });
} else if (level === 10) {
  process.on('workerMessage', (value, source) => {
    console.log(`${source} -> ${threadId}:`, value);
    channel.postMessage('done');
    channel.close();
  });

  await postMessageToThread(0, { message: 'ping' });
}

channel.onmessage = channel.close;

const {
  postMessageToThread,
  threadId,
  workerData,
  Worker,
} = require('node:worker_threads');

const channel = new BroadcastChannel('sync');
const level = workerData?.level ?? 0;

if (level < 10) {
  const worker = new Worker(__filename, {
    workerData: { level: level + 1 },
  });
}

if (level === 0) {
  process.on('workerMessage', (value, source) => {
    console.log(`${source} -> ${threadId}:`, value);
    postMessageToThread(source, { message: 'pong' });
  });
} else if (level === 10) {
  process.on('workerMessage', (value, source) => {
    console.log(`${source} -> ${threadId}:`, value);
    channel.postMessage('done');
    channel.close();
  });

  postMessageToThread(0, { message: 'ping' });
}

channel.onmessage = channel.close;

worker_threads.receiveMessageOnPort(port)

• port <MessagePort> | <BroadcastChannel>

• Возвращает: <Object> | <undefined>

Получает одно сообщение из указанного MessagePort. Если сообщений нет, возвращается undefined; в противном случае возвращается объект с единственным свойством message, содержащим полезную нагрузку сообщения, соответствующего самому старому сообщению в очереди MessagePort.

import { MessageChannel, receiveMessageOnPort } from 'node:worker_threads';
const { port1, port2 } = new MessageChannel();
port1.postMessage({ hello: 'world' });

console.log(receiveMessageOnPort(port2));
// Prints: { message: { hello: 'world' } }
console.log(receiveMessageOnPort(port2));
// Prints: undefined

const { MessageChannel, receiveMessageOnPort } = require('node:worker_threads');
const { port1, port2 } = new MessageChannel();
port1.postMessage({ hello: 'world' });

console.log(receiveMessageOnPort(port2));
// Prints: { message: { hello: 'world' } }
console.log(receiveMessageOnPort(port2));
// Prints: undefined

При использовании этой функции событие 'message' не генерируется, а обработчик onmessage не вызывается.

worker_threads.resourceLimits

• Тип: <Object>
  • maxYoungGenerationSizeMb <number>
  • maxOldGenerationSizeMb <number>
  • codeRangeSizeMb <number>
  • stackSizeMb <number>

Предоставляет набор ограничений ресурсов движка JS для этого Worker-потока. Если конструктору Worker был передан параметр resourceLimits, значения будут соответствовать ему.

При использовании в главном потоке возвращается пустой объект.

worker_threads.SHARE_ENV

• Тип: <symbol>

Специальное значение, которое можно передать в качестве параметра env конструктора Worker, чтобы указать, что текущий поток и Worker-поток должны совместно использовать один и тот же набор переменных окружения с доступом на чтение и запись.

import process from 'node:process';
import { Worker, SHARE_ENV } from 'node:worker_threads';
new Worker('process.env.SET_IN_WORKER = "foo"', { eval: true, env: SHARE_ENV })
  .once('exit', () => {
    console.log(process.env.SET_IN_WORKER);  // Prints 'foo'.
  });

const { Worker, SHARE_ENV } = require('node:worker_threads');
new Worker('process.env.SET_IN_WORKER = "foo"', { eval: true, env: SHARE_ENV })
  .once('exit', () => {
    console.log(process.env.SET_IN_WORKER);  // Prints 'foo'.
  });

worker_threads.setEnvironmentData(key[, value])

• key <any> Любое произвольное клонируемое значение JavaScript, которое можно использовать в качестве ключа <Map>.
• value <any> Любое произвольное клонируемое значение JavaScript, которое будет клонировано и автоматически передано всем новым экземплярам Worker. Если в качестве undefined передано value, ранее заданное значение для key будет удалено.

API worker.setEnvironmentData() задаёт содержимое worker.getEnvironmentData() в текущем потоке и во всех новых экземплярах Worker, созданных из текущего контекста.

worker_threads.threadId

• Тип: <integer>

Целочисленный идентификатор текущего потока. В соответствующем объекте worker (если он есть) он доступен как worker.threadId. Это значение уникально для каждого экземпляра Worker в рамках одного процесса.

worker_threads.threadName

• <string> | <null>

Строковый идентификатор текущего потока или null, если поток не запущен. В соответствующем объекте worker (если он есть) он доступен как worker.threadName.

worker_threads.workerData

Произвольное значение JavaScript, содержащее копию данных, переданных конструктору Worker этого потока.

Данные клонируются так же, как при использовании postMessage(), в соответствии с алгоритмом структурированного клонирования HTML.

import { Worker, isMainThread, workerData } from 'node:worker_threads';

if (isMainThread) {
  const worker = new Worker(new URL(import.meta.url), { workerData: 'Hello, world!' });
} else {
  console.log(workerData);  // Prints 'Hello, world!'.
}

const { Worker, isMainThread, workerData } = require('node:worker_threads');

if (isMainThread) {
  const worker = new Worker(__filename, { workerData: 'Hello, world!' });
} else {
  console.log(workerData);  // Prints 'Hello, world!'.
}

worker_threads.locks

• <LockManager>

Экземпляр LockManager, который можно использовать для координации доступа к ресурсам, совместно используемым несколькими потоками в одном процессе. Семантика API соответствует семантике LockManager браузера.

Класс: Lock

Интерфейс Lock предоставляет сведения о блокировке, полученной через locks.request().

lock.name

• <string>

Имя блокировки.

lock.mode

• <string>

Режим блокировки: shared или exclusive.

Класс: LockManager

Интерфейс LockManager предоставляет методы для запроса блокировок и получения сведений о них. Чтобы получить экземпляр LockManager, используйте

import { locks } from 'node:worker_threads';

const { locks } = require('node:worker_threads');

Эта реализация соответствует API LockManager браузера.

locks.request(name[, options], callback)

• name <string>
• options <Object>
  • mode <string> 'exclusive' или 'shared'. По умолчанию: 'exclusive'.
  • ifAvailable <boolean> Если true, запрос будет удовлетворён, только если блокировка ещё не занята. Если запрос нельзя удовлетворить, будет вызван callback со значением null вместо экземпляра Lock. По умолчанию: false.
  • steal <boolean> Если true, все существующие блокировки с тем же именем освобождаются, а запрос удовлетворяется немедленно, опережая запросы в очереди. По умолчанию: false.
  • signal <AbortSignal>, который можно использовать для отмены ожидающего запроса блокировки (ещё не удовлетворённого).
• callback <Function> Вызывается после предоставления блокировки (или немедленно со значением null, если ifAvailable равно true и блокировка недоступна). Блокировка автоматически освобождается при возврате функции или, если функция возвращает промис, после его завершения.
• Возвращает: <Promise> Выполняется после освобождения блокировки.

import { locks } from 'node:worker_threads';

await locks.request('my_resource', async (lock) => {
  // The lock has been acquired.
});
// The lock has been released here.

const { locks } = require('node:worker_threads');

locks.request('my_resource', async (lock) => {
  // The lock has been acquired.
}).then(() => {
  // The lock has been released here.
});

locks.query()

• Возвращает: <Promise>

Выполняется с объектом LockManagerSnapshot, описывающим текущие занятые и ожидающие блокировки процесса.

import { locks } from 'node:worker_threads';

const snapshot = await locks.query();
for (const lock of snapshot.held) {
  console.log(`held lock: name ${lock.name}, mode ${lock.mode}`);
}
for (const pending of snapshot.pending) {
  console.log(`pending lock: name ${pending.name}, mode ${pending.mode}`);
}

const { locks } = require('node:worker_threads');

locks.query().then((snapshot) => {
  for (const lock of snapshot.held) {
    console.log(`held lock: name ${lock.name}, mode ${lock.mode}`);
  }
  for (const pending of snapshot.pending) {
    console.log(`pending lock: name ${pending.name}, mode ${pending.mode}`);
  }
});

Класс: BroadcastChannel extends EventTarget

Экземпляры BroadcastChannel обеспечивают асинхронную связь «один ко многим» со всеми остальными экземплярами BroadcastChannel, подключёнными к каналу с тем же именем.

import {
  isMainThread,
  BroadcastChannel,
  Worker,
} from 'node:worker_threads';

const bc = new BroadcastChannel('hello');

if (isMainThread) {
  let c = 0;
  bc.onmessage = (event) => {
    console.log(event.data);
    if (++c === 10) bc.close();
  };
  for (let n = 0; n < 10; n++)
    new Worker(new URL(import.meta.url));
} else {
  bc.postMessage('hello from every worker');
  bc.close();
}

const {
  isMainThread,
  BroadcastChannel,
  Worker,
} = require('node:worker_threads');

const bc = new BroadcastChannel('hello');

if (isMainThread) {
  let c = 0;
  bc.onmessage = (event) => {
    console.log(event.data);
    if (++c === 10) bc.close();
  };
  for (let n = 0; n < 10; n++)
    new Worker(__filename);
} else {
  bc.postMessage('hello from every worker');
  bc.close();
}

new BroadcastChannel(name)

• name <any> Имя канала, к которому нужно подключиться. Допускается любое значение JavaScript, которое можно преобразовать в строку с помощью `${name}`.

broadcastChannel.close()

Закрывает соединение BroadcastChannel.

broadcastChannel.onmessage

• Тип: <Function> Вызывается с одним аргументом MessageEvent при получении сообщения.

broadcastChannel.onmessageerror

• Тип: <Function> Вызывается, если полученное сообщение не удаётся десериализовать.

broadcastChannel.postMessage(message)

• message <any> Любое клонируемое значение JavaScript.

broadcastChannel.ref()

Действие, противоположное unref(). Вызов ref() для ранее unref()-нутого BroadcastChannel не позволяет программе завершиться, если это единственный оставшийся активный дескриптор (поведение по умолчанию). Если порт ref()-нут, повторный вызов ref() ни на что не повлияет.

broadcastChannel.unref()

Вызов unref() для BroadcastChannel позволяет потоку завершиться, если это единственный активный дескриптор в системе событий. Если BroadcastChannel уже unref()-нут, повторный вызов unref() ни на что не повлияет.

Класс: MessageChannel

Экземпляры класса worker.MessageChannel представляют собой асинхронный двусторонний канал связи. У MessageChannel нет собственных методов. new MessageChannel() создаёт объект со свойствами port1 и port2, которые ссылаются на связанные экземпляры MessagePort.

import { MessageChannel } from 'node:worker_threads';

const { port1, port2 } = new MessageChannel();
port1.on('message', (message) => console.log('received', message));
port2.postMessage({ foo: 'bar' });
// Prints: received { foo: 'bar' } from the `port1.on('message')` listener

const { MessageChannel } = require('node:worker_threads');

const { port1, port2 } = new MessageChannel();
port1.on('message', (message) => console.log('received', message));
port2.postMessage({ foo: 'bar' });
// Prints: received { foo: 'bar' } from the `port1.on('message')` listener

Класс: MessagePort

• Расширяет: <EventTarget>

Экземпляры класса worker.MessagePort представляют один конец асинхронного двустороннего канала связи. Его можно использовать для передачи структурированных данных, областей памяти и других объектов MessagePort между разными экземплярами Worker.

Эта реализация соответствует MessagePort браузера.

Событие: 'close'

Событие 'close' генерируется, когда любая из сторон канала отключается.

import { MessageChannel } from 'node:worker_threads';
const { port1, port2 } = new MessageChannel();

// Prints:
//   foobar
//   closed!
port2.on('message', (message) => console.log(message));
port2.once('close', () => console.log('closed!'));

port1.postMessage('foobar');
port1.close();

const { MessageChannel } = require('node:worker_threads');
const { port1, port2 } = new MessageChannel();

// Prints:
//   foobar
//   closed!
port2.on('message', (message) => console.log(message));
port2.once('close', () => console.log('closed!'));

port1.postMessage('foobar');
port1.close();

Событие: 'message'

• value <any> Передаваемое значение

Событие 'message' генерируется для каждого входящего сообщения, содержащего клонированные входные данные метода port.postMessage().

Обработчики этого события получают копию параметра value, переданного в postMessage(), без каких-либо дополнительных аргументов.

Событие: 'messageerror'

• error <Error> Объект Error

Событие 'messageerror' генерируется при ошибке десериализации сообщения.

В настоящее время это событие генерируется, если при создании отправленного объекта JS на принимающей стороне возникает ошибка. Такие ситуации редки, но возможны, например, когда определённые объекты API Node.js получены в vm.Context (где API Node.js в настоящее время недоступны).

port.close()

Отключает дальнейшую отправку сообщений с обеих сторон соединения. Этот метод можно вызвать, если дальнейшее взаимодействие через этот MessagePort не предполагается.

Событие 'close' генерируется для обоих экземпляров MessagePort, участвующих в канале.

port.postMessage(value[, transferList])

• value <any>
• transferList <Object[]>

Отправляет значение JavaScript принимающей стороне этого канала. value передаётся способом, совместимым с алгоритмом структурированного клонирования HTML.

В частности, существенные отличия от JSON таковы:

• value может содержать циклические ссылки.
• value может содержать экземпляры встроенных типов JS, например RegExps, BigInts, Maps, Sets и т. д.
• value может содержать типизированные массивы, использующие как ArrayBuffers, так и SharedArrayBuffers.
• value может содержать экземпляры WebAssembly.Module.
• value не может содержать нативные объекты (на основе C++), кроме следующих:
  • <CryptoKey>s,
  • <FileHandle>s,
  • <Histogram>s,
  • <KeyObject>s,
  • <MessagePort>s,
  • <net.BlockList>s,
  • <net.SocketAddress>es,
  • <X509Certificate>s.

import { MessageChannel } from 'node:worker_threads';
const { port1, port2 } = new MessageChannel();

port1.on('message', (message) => console.log(message));

const circularData = {};
circularData.foo = circularData;
// Prints: { foo: [Circular] }
port2.postMessage(circularData);

const { MessageChannel } = require('node:worker_threads');
const { port1, port2 } = new MessageChannel();

port1.on('message', (message) => console.log(message));

const circularData = {};
circularData.foo = circularData;
// Prints: { foo: [Circular] }
port2.postMessage(circularData);

transferList может быть списком объектов <ArrayBuffer>, MessagePort и FileHandle. После передачи их нельзя использовать на отправляющей стороне канала (даже если они не входят в value). В отличие от дочерних процессов, передача дескрипторов, например сетевых сокетов, в настоящее время не поддерживается.

Если value содержит экземпляры <SharedArrayBuffer>, они доступны из обоих потоков. Их нельзя указывать в transferList.

value может по-прежнему содержать экземпляры ArrayBuffer, отсутствующие в transferList; в этом случае базовая память копируется, а не перемещается.

import { MessageChannel } from 'node:worker_threads';
const { port1, port2 } = new MessageChannel();

port1.on('message', (message) => console.log(message));

const uint8Array = new Uint8Array([ 1, 2, 3, 4 ]);
// This posts a copy of `uint8Array`:
port2.postMessage(uint8Array);
// This does not copy data, but renders `uint8Array` unusable:
port2.postMessage(uint8Array, [ uint8Array.buffer ]);

// The memory for the `sharedUint8Array` is accessible from both the
// original and the copy received by `.on('message')`:
const sharedUint8Array = new Uint8Array(new SharedArrayBuffer(4));
port2.postMessage(sharedUint8Array);

// This transfers a freshly created message port to the receiver.
// This can be used, for example, to create communication channels between
// multiple `Worker` threads that are children of the same parent thread.
const otherChannel = new MessageChannel();
port2.postMessage({ port: otherChannel.port1 }, [ otherChannel.port1 ]);

const { MessageChannel } = require('node:worker_threads');
const { port1, port2 } = new MessageChannel();

port1.on('message', (message) => console.log(message));

const uint8Array = new Uint8Array([ 1, 2, 3, 4 ]);
// This posts a copy of `uint8Array`:
port2.postMessage(uint8Array);
// This does not copy data, but renders `uint8Array` unusable:
port2.postMessage(uint8Array, [ uint8Array.buffer ]);

// The memory for the `sharedUint8Array` is accessible from both the
// original and the copy received by `.on('message')`:
const sharedUint8Array = new Uint8Array(new SharedArrayBuffer(4));
port2.postMessage(sharedUint8Array);

// This transfers a freshly created message port to the receiver.
// This can be used, for example, to create communication channels between
// multiple `Worker` threads that are children of the same parent thread.
const otherChannel = new MessageChannel();
port2.postMessage({ port: otherChannel.port1 }, [ otherChannel.port1 ]);

Объект сообщения клонируется немедленно, и его можно изменять после отправки без побочных эффектов.

Дополнительные сведения о механизмах сериализации и десериализации, лежащих в основе этого API, см. в API сериализации модуля node:v8.

Особенности передачи типизированных массивов и буферов

Все экземпляры <TypedArray> | <Buffer> являются представлениями базового <ArrayBuffer>. Иными словами, именно ArrayBuffer хранит исходные данные, тогда как объекты TypedArray и Buffer предоставляют способ просмотра данных и управления ими. Над одним экземпляром ArrayBuffer можно создать несколько представлений, и это часто делается. При использовании списка передачи для передачи ArrayBuffer следует проявлять особую осторожность, поскольку в результате все экземпляры TypedArray и Buffer, использующие тот же ArrayBuffer, станут непригодными для использования.

const ab = new ArrayBuffer(10);

const u1 = new Uint8Array(ab);
const u2 = new Uint16Array(ab);

console.log(u2.length);  // prints 5

port.postMessage(u1, [u1.buffer]);

console.log(u2.length);  // prints 0
jscopy

В частности, для экземпляров Buffer возможность передачи или клонирования базового ArrayBuffer полностью зависит от способа создания экземпляров, который часто невозможно надёжно определить.

Для ArrayBuffer можно вызвать markAsUntransferable(), чтобы указать, что его всегда следует клонировать, а не передавать.

В зависимости от способа создания экземпляра Buffer он может владеть базовым ArrayBuffer или не владеть им. ArrayBuffer нельзя передавать, если не известно, что экземпляр Buffer владеет им. В частности, экземпляры Buffer, созданные из внутреннего пула Buffer (например, с помощью Buffer.from() или Buffer.allocUnsafe()), передать нельзя: они всегда клонируются, что отправляет копию всего пула Buffer. Такое поведение может привести к непреднамеренному увеличению использования памяти и потенциальным проблемам безопасности.

Дополнительные сведения об объединении Buffer в пулы см. в разделе Buffer.allocUnsafe().

Объекты ArrayBuffer экземпляров Buffer, созданных с помощью Buffer.alloc() или Buffer.allocUnsafeSlow(), всегда можно передавать, однако при этом все остальные существующие представления этих ArrayBuffer станут непригодными для использования.

Особенности клонирования объектов с прототипами, классами и методами доступа

Поскольку при клонировании объектов используется алгоритм структурированного клонирования HTML, неперечисляемые свойства, методы доступа к свойствам и прототипы объектов не сохраняются. В частности, объекты <Buffer> на принимающей стороне будут распознаны как обычные <Uint8Array>, а экземпляры классов JavaScript будут клонированы как обычные объекты JavaScript.

const b = Symbol('b');

class Foo {
  #a = 1;
  constructor() {
    this[b] = 2;
    this.c = 3;
  }

  get d() { return this.#a + 3; }
}

const { port1, port2 } = new MessageChannel();

port1.onmessage = ({ data }) => console.log(data);

port2.postMessage(new Foo());

// Prints: { c: 3 }
jscopy

Некоторые встроенные объекты вообще нельзя клонировать. Например, отправка объекта URL вызывает ошибку DataCloneError:

const { port1, port2 } = new MessageChannel();

port2.postMessage(new URL('https://example.org'));
// Throws DataCloneError: Cannot clone object of unsupported type.
jscopy

port.hasRef()

• Возвращает: <boolean>

Если значение равно true, объект MessagePort не позволит циклу событий Node.js завершиться.

port.ref()

Противоположность unref(). Вызов ref() для порта, для которого ранее был вызван unref(), не позволит программе завершиться, если это единственный оставшийся активный дескриптор (поведение по умолчанию). Если для порта вызван ref(), повторный вызов ref() не даст эффекта.

Если обработчики событий добавляются или удаляются с помощью .on('message'), для порта автоматически вызываются ref() и unref() в зависимости от наличия обработчиков события.

port.start()

Начинает получать сообщения через этот MessagePort. Если этот порт используется как генератор событий, метод вызывается автоматически после добавления обработчиков 'message'.

Этот метод существует для совместимости с API MessagePort Web. В Node.js он полезен только для игнорирования сообщений, когда обработчик события отсутствует. Обработка .onmessage в Node.js также отличается. При установке этого значения автоматически вызывается .start(), но при его сбросе сообщения накапливаются в очереди до установки нового обработчика или удаления порта.

port.unref()

Вызов unref() для порта позволяет потоку завершиться, если этот порт является единственным активным дескриптором в системе событий. Если для порта уже вызван unref(), повторный вызов unref() не даст эффекта.

Если обработчики событий добавляются или удаляются с помощью .on('message'), для порта автоматически вызываются ref() и unref() в зависимости от наличия обработчиков события.

Класс: Worker

• Расширяет: <EventEmitter>

Класс Worker представляет независимый поток выполнения JavaScript. В нём доступны большинство API Node.js.

В среде Worker есть следующие существенные отличия:

• Потоки process.stdin, process.stdout и process.stderr могут перенаправляться родительским потоком.
• Свойству require('node:worker_threads').isMainThread присваивается значение false.
• Доступен порт сообщений require('node:worker_threads').parentPort.
• process.exit() останавливает не всю программу, а только отдельный поток; process.abort() недоступен.
• Методы process.chdir() и process, задающие идентификаторы группы или пользователя, недоступны.
• process.env является копией переменных окружения родительского потока, если не указано иное. Изменения одной копии не видны в других потоках и нативных дополнениях (если только worker.SHARE_ENV не передан как параметр env конструктору Worker). В Windows, в отличие от главного потока, копия переменных окружения чувствительна к регистру.
• process.title нельзя изменить.
• Сигналы не передаются через process.on('...').
• Выполнение может остановиться в любой момент в результате вызова worker.terminate().
• Канал IPC родительских процессов недоступен.
• Модуль trace_events не поддерживается.
• Нативные дополнения можно загружать из нескольких потоков, только если они соответствуют определённым условиям.

Можно создавать экземпляры Worker внутри других экземпляров Worker.

Как и в Web Workers и модуле node:cluster, двустороннее взаимодействие можно организовать с помощью обмена сообщениями между потоками. Внутри Worker содержит встроенную пару экземпляров MessagePort, уже связанных друг с другом при создании Worker. Хотя объект MessagePort на стороне родительского потока напрямую не предоставляется, его функции доступны через worker.postMessage() и событие worker.on('message') объекта Worker родительского потока.

Чтобы создавать собственные каналы обмена сообщениями (это предпочтительнее использования глобального канала по умолчанию, поскольку помогает разделять области ответственности), пользователи могут создать объект MessageChannel в любом потоке и передать один из его экземпляров MessagePort из этого MessageChannel в другой поток через уже существующий канал, например глобальный.

Дополнительные сведения о передаче сообщений и о том, какие значения JavaScript можно успешно передавать между потоками, см. в разделе port.postMessage().

import assert from 'node:assert';
import {
  Worker, MessageChannel, MessagePort, isMainThread, parentPort,
} from 'node:worker_threads';
if (isMainThread) {
  const worker = new Worker(new URL(import.meta.url));
  const subChannel = new MessageChannel();
  worker.postMessage({ hereIsYourPort: subChannel.port1 }, [subChannel.port1]);
  subChannel.port2.on('message', (value) => {
    console.log('received:', value);
  });
} else {
  parentPort.once('message', (value) => {
    assert(value.hereIsYourPort instanceof MessagePort);
    value.hereIsYourPort.postMessage('the worker is sending this');
    value.hereIsYourPort.close();
  });
}

const assert = require('node:assert');
const {
  Worker, MessageChannel, MessagePort, isMainThread, parentPort,
} = require('node:worker_threads');
if (isMainThread) {
  const worker = new Worker(__filename);
  const subChannel = new MessageChannel();
  worker.postMessage({ hereIsYourPort: subChannel.port1 }, [subChannel.port1]);
  subChannel.port2.on('message', (value) => {
    console.log('received:', value);
  });
} else {
  parentPort.once('message', (value) => {
    assert(value.hereIsYourPort instanceof MessagePort);
    value.hereIsYourPort.postMessage('the worker is sending this');
    value.hereIsYourPort.close();
  });
}

new Worker(filename[, options])

• filename <string> | <URL> Путь к основному скрипту или модулю Worker. Должен быть абсолютным путём или относительным путём (то есть относительно текущего рабочего каталога), начинающимся с ./ или ../, либо объектом WHATWG URL, использующим протокол file: или data:. При использовании URL data: данные интерпретируются на основе MIME-типа с помощью загрузчика модулей ECMAScript. Если options.eval имеет значение true, это строка с кодом JavaScript, а не путь.
• options <Object>
  • argv <any[]> Список аргументов, которые будут преобразованы в строки и добавлены к process.argv в Worker. В основном это аналог workerData, однако значения доступны в глобальном process.argv, как если бы они были переданы скрипту в качестве параметров CLI.
  • env <Object> Если задано, определяет начальное значение process.env в потоке Worker. В качестве специального значения можно использовать worker.SHARE_ENV, чтобы указать, что родительский и дочерний потоки должны совместно использовать переменные окружения; в этом случае изменения объекта process.env одного потока также влияют на другой поток. По умолчанию: process.env.
  • eval <boolean> Если установлено значение true и первый аргумент имеет тип string, первый аргумент конструктора интерпретируется как скрипт, который выполняется после запуска Worker.
  • execArgv <string[]> Список параметров CLI Node.js, передаваемых Worker. Параметры V8 (например, --max-old-space-size) и параметры, влияющие на процесс (например, --title), не поддерживаются. Если задано, значение доступно в Worker через process.execArgv. По умолчанию параметры наследуются от родительского потока.
  • stdin <boolean> Если установлено значение true, worker.stdin предоставляет поток для записи, содержимое которого появляется в Worker как process.stdin. По умолчанию данные не предоставляются.
  • stdout <boolean> Если установлено значение true, worker.stdout не перенаправляется автоматически в process.stdout родительского потока.
  • stderr <boolean> Если установлено значение true, worker.stderr не перенаправляется автоматически в process.stderr родительского потока.
  • workerData <any> Любое значение JavaScript, которое клонируется и становится доступным как require('node:worker_threads').workerData. Клонирование выполняется, как описано в алгоритме структурированного клонирования HTML; если объект нельзя клонировать (например, из-за содержащихся в нём functions), генерируется ошибка.
  • trackUnmanagedFds <boolean> Если установлено значение true, Worker отслеживает необработанные файловые дескрипторы, управляемые с помощью fs.open() и fs.close(), и закрывает их при завершении работы Worker, как и другие ресурсы, например сетевые сокеты или файловые дескрипторы, управляемые через API FileHandle. Этот параметр автоматически наследуется всеми вложенными экземплярами Worker. По умолчанию: true.
  • transferList <Object[]> Если в workerData передан один или несколько объектов типа MessagePort, для этих элементов требуется transferList, иначе генерируется ошибка ERR_MISSING_MESSAGE_PORT_IN_TRANSFER_LIST. Дополнительные сведения см. в разделе port.postMessage().
  • resourceLimits <Object> Необязательный набор ограничений ресурсов для нового экземпляра движка JS. При достижении этих ограничений экземпляр Worker завершается. Эти ограничения влияют только на движок JS, но не на внешние данные, в том числе на ArrayBuffers. Даже при заданных ограничениях процесс всё ещё может аварийно завершиться при глобальной нехватке памяти.
    • maxOldGenerationSizeMb <number> Максимальный размер основной кучи в МБ. Если задан аргумент командной строки --max-old-space-size, он переопределяет это значение.
    • maxYoungGenerationSizeMb <number> Максимальный размер области кучи для недавно созданных объектов. Если задан аргумент командной строки --max-semi-space-size, он переопределяет это значение.
    • codeRangeSizeMb <number> Размер предварительно выделенного диапазона памяти для сгенерированного кода.
    • stackSizeMb <number> Максимальный размер стека потока по умолчанию. Малые значения могут привести к невозможности использования экземпляров Worker. По умолчанию: 4.
  • name <string> Необязательное name, которое будет подставлено в имя потока и заголовок Worker для отладки и идентификации; итоговый заголовок будет иметь вид [worker ${id}] ${name}. Максимально допустимый размер этого параметра зависит от операционной системы. Если указанное имя превышает ограничение, оно будет усечено
    • Максимальные размеры:
      • Windows: 32 767 символов
      • macOS: 64 символа
      • Linux: 16 символов
      • NetBSD: ограничено значением PTHREAD_MAX_NAMELEN_NP
      • FreeBSD и OpenBSD: ограничено значением MAXCOMLEN По умолчанию: 'WorkerThread'.

Событие: 'error'

• err <any>

Событие 'error' генерируется, если в потоке Worker возникает необработанное исключение. В этом случае Worker завершается.

Событие: 'exit'

• exitCode <integer>

Событие 'exit' генерируется после остановки Worker. Если Worker завершился вызовом process.exit(), параметр exitCode содержит переданный код завершения. Если Worker был принудительно завершён, параметр exitCode имеет значение 1.

Это последнее событие, генерируемое любым экземпляром Worker.

Событие: 'message'

• value <any> Передаваемое значение

Событие 'message' генерируется, когда поток Worker вызывает require('node:worker_threads').parentPort.postMessage(). Дополнительные сведения см. в описании события port.on('message').

Все сообщения, отправленные из потока Worker, генерируются до события 'exit' для объекта Worker.

Событие: 'messageerror'

• error <Error> Объект Error

Событие 'messageerror' генерируется при ошибке десериализации сообщения.

Событие: 'online'

Событие 'online' генерируется, когда поток Worker начинает выполнять код JavaScript.

worker.cpuUsage([prev])

• Возвращает: <Promise>

Этот метод возвращает Promise, который разрешается объектом, идентичным результату process.threadCpuUsage(), или отклоняется с ошибкой ERR_WORKER_NOT_RUNNING, если Worker больше не работает. Этот метод позволяет просматривать статистику извне самого потока.

worker.getHeapSnapshot([options])

• options <Object>
  • exposeInternals <boolean> Если значение равно true, во время создания снимка кучи становятся видны внутренние данные. По умолчанию: false.
  • exposeNumericValues <boolean> Если значение равно true, числовые значения становятся видны в искусственных полях. По умолчанию: false.
• Возвращает: <Promise> Promise для потока Readable, содержащего снимок кучи V8

Возвращает поток для чтения, содержащий снимок текущего состояния Worker в V8. Подробнее см. в разделе v8.getHeapSnapshot().

Если поток Worker больше не выполняется, что может произойти до генерации 'exit' события, возвращённый Promise немедленно отклоняется с ошибкой ERR_WORKER_NOT_RUNNING.

worker.getHeapStatistics()

• Возвращает: <Promise>

Этот метод возвращает Promise, который разрешается объектом, идентичным результату v8.getHeapStatistics(), или отклоняется с ошибкой ERR_WORKER_NOT_RUNNING, если Worker больше не выполняется. Этот метод позволяет получать статистику из-за пределов самого потока.

worker.performance

Объект, который можно использовать для получения сведений о производительности экземпляра Worker.

performance.eventLoopUtilization([utilization1[, utilization2]])

• utilization1 <Object> Результат предыдущего вызова eventLoopUtilization().
• utilization2 <Object> Результат предыдущего вызова eventLoopUtilization(), выполненного до utilization1.
• Возвращает: <Object>
  • idle <number>
  • active <number>
  • utilization <number>

Это тот же вызов, что и perf_hooks eventLoopUtilization(), за исключением того, что возвращаются значения экземпляра Worker.

Одно из отличий заключается в том, что, в отличие от основного потока, загрузка Worker выполняется в цикле событий. Поэтому данные об использовании цикла событий доступны сразу после начала выполнения скрипта Worker.

Неизменяющееся значение времени idle не означает, что Worker завис во время загрузки. В следующих примерах показано, что за всё время работы Worker не накапливается время idle, но он всё равно может обрабатывать сообщения.

import { Worker, isMainThread, parentPort } from 'node:worker_threads';

if (isMainThread) {
  const worker = new Worker(new URL(import.meta.url));
  setInterval(() => {
    worker.postMessage('hi');
    console.log(worker.performance.eventLoopUtilization());
  }, 100).unref();
} else {
  parentPort.on('message', () => console.log('msg')).unref();
  (function r(n) {
    if (--n < 0) return;
    const t = Date.now();
    while (Date.now() - t < 300);
    setImmediate(r, n);
  })(10);
}

const { Worker, isMainThread, parentPort } = require('node:worker_threads');

if (isMainThread) {
  const worker = new Worker(__filename);
  setInterval(() => {
    worker.postMessage('hi');
    console.log(worker.performance.eventLoopUtilization());
  }, 100).unref();
} else {
  parentPort.on('message', () => console.log('msg')).unref();
  (function r(n) {
    if (--n < 0) return;
    const t = Date.now();
    while (Date.now() - t < 300);
    setImmediate(r, n);
  })(10);
}

Данные об использовании цикла событий Worker доступны только после генерации 'online' события. Если вызвать метод до этого момента или после генерации 'exit' события, все свойства будут иметь значение 0.

worker.postMessage(value[, transferList])

• value <any>
• transferList <Object[]>

Отправляет Worker сообщение, которое принимается через require('node:worker_threads').parentPort.on('message'). Подробнее см. в разделе port.postMessage().

worker.ref()

Противоположность unref(): вызов ref() для Worker, к которому ранее был применён unref(), не позволяет программе завершить работу, если он является единственным оставшимся активным дескриптором (поведение по умолчанию). Если к Worker применён ref(), повторный вызов ref() не оказывает эффекта.

worker.resourceLimits

• Тип: <Object>
  • maxYoungGenerationSizeMb <number>
  • maxOldGenerationSizeMb <number>
  • codeRangeSizeMb <number>
  • stackSizeMb <number>

Предоставляет набор ограничений ресурсов движка JS для этого потока Worker. Если при создании Worker был передан параметр resourceLimits, его значения совпадают со значениями этого параметра.

Если Worker остановлен, возвращается пустой объект.

worker.startCpuProfile([options])

• options <Object>
  • sampleInterval <number> Запрошенный интервал выборки в миллисекундах. По умолчанию: 0.
  • maxBufferSize <integer> Максимальное количество сохраняемых образцов. По умолчанию: 4294967295.
• Возвращает: <Promise>

Запускает профилирование ЦП, а затем возвращает Promise, который разрешается ошибкой или объектом CPUProfileHandle. Этот API поддерживает синтаксис await using.

const { Worker } = require('node:worker_threads');

const worker = new Worker(`
  const { parentPort } = require('worker_threads');
  parentPort.on('message', () => {});
  `, { eval: true });

worker.on('online', async () => {
  const handle = await worker.startCpuProfile({ sampleInterval: 1 });
  const profile = await handle.stop();
  console.log(profile);
  worker.terminate();
});
cjscopy

Пример await using.

const { Worker } = require('node:worker_threads');

const w = new Worker(`
  const { parentPort } = require('node:worker_threads');
  parentPort.on('message', () => {});
  `, { eval: true });

w.on('online', async () => {
  // Stop profile automatically when return and profile will be discarded
  await using handle = await w.startCpuProfile();
});
cjscopy

worker.startHeapProfile([options])

• options <Object>
  • sampleInterval <number> Средний интервал выборки в байтах. По умолчанию: 524288 (512 КиБ).
  • stackDepth <integer> Максимальная глубина стека для образцов. По умолчанию: 16.
  • forceGC <boolean> Принудительно выполнить сборку мусора перед созданием профиля. По умолчанию: false.
  • includeObjectsCollectedByMajorGC <boolean> Включать объекты, собранные при полной сборке мусора. По умолчанию: false.
  • includeObjectsCollectedByMinorGC <boolean> Включать объекты, собранные при частичной сборке мусора. По умолчанию: false.
• Возвращает: <Promise>

Запускает профилирование кучи, а затем возвращает Promise, который разрешается ошибкой или объектом HeapProfileHandle. Этот API поддерживает синтаксис await using.

const { Worker } = require('node:worker_threads');

const worker = new Worker(`
  const { parentPort } = require('worker_threads');
  parentPort.on('message', () => {});
  `, { eval: true });

worker.on('online', async () => {
  const handle = await worker.startHeapProfile();
  const profile = await handle.stop();
  console.log(profile);
  worker.terminate();
});

import { Worker } from 'node:worker_threads';

const worker = new Worker(`
  const { parentPort } = require('node:worker_threads');
  parentPort.on('message', () => {});
  `, { eval: true });

worker.on('online', async () => {
  const handle = await worker.startHeapProfile();
  const profile = await handle.stop();
  console.log(profile);
  worker.terminate();
});

Пример await using.

const { Worker } = require('node:worker_threads');

const w = new Worker(`
  const { parentPort } = require('node:worker_threads');
  parentPort.on('message', () => {});
  `, { eval: true });

w.on('online', async () => {
  // Stop profile automatically when return and profile will be discarded
  await using handle = await w.startHeapProfile();
});

import { Worker } from 'node:worker_threads';

const w = new Worker(`
  const { parentPort } = require('node:worker_threads');
  parentPort.on('message', () => {});
  `, { eval: true });

w.on('online', async () => {
  // Stop profile automatically when return and profile will be discarded
  await using handle = await w.startHeapProfile();
});

worker.stderr

• Тип: <stream.Readable>

Это поток для чтения, содержащий данные, записанные в process.stderr внутри потока Worker. Если при создании Worker не был передан параметр stderr: true, данные перенаправляются в поток process.stderr родительского потока.

worker.stdin

• Тип: <null> | <stream.Writable>

Если при создании Worker был передан параметр stdin: true, это поток для записи. Данные, записанные в этот поток, становятся доступны в потоке Worker как process.stdin.

worker.stdout

• Тип: <stream.Readable>

Это поток для чтения, содержащий данные, записанные в process.stdout внутри потока Worker. Если при создании Worker не был передан параметр stdout: true, данные перенаправляются в поток process.stdout родительского потока.

worker.terminate()

• Возвращает: <Promise>

Как можно скорее останавливает выполнение всего кода JavaScript в потоке Worker. Возвращает Promise с кодом завершения, который разрешается при генерации 'exit' события.

worker.threadId

• Тип: <integer>

Целочисленный идентификатор указанного потока. Внутри потока Worker он доступен как require('node:worker_threads').threadId. Это значение уникально для каждого экземпляра Worker в пределах одного процесса.

worker.threadName

• <string> | <null>

Строковый идентификатор указанного потока или null, если поток не выполняется. Внутри потока Worker он доступен как require('node:worker_threads').threadName.

worker.unref()

Вызов unref() для Worker позволяет потоку завершить работу, если он является единственным активным дескриптором в системе событий. Если к Worker уже применён unref(), повторный вызов unref() не оказывает эффекта.

worker[Symbol.asyncDispose]()

Вызывает worker.terminate() при выходе из области видимости очистки.

async function example() {
  await using worker = new Worker('for (;;) {}', { eval: true });
  // Worker is automatically terminate when the scope is exited.
}
jscopy

Примечания

Синхронная блокировка стандартных потоков ввода-вывода

Workerы используют передачу сообщений через <MessagePort> для взаимодействия с stdio. Это означает, что вывод stdio, исходящий от Worker, может быть заблокирован синхронным кодом на стороне получателя, который блокирует цикл событий Node.js.

import {
  Worker,
  isMainThread,
} from 'node:worker_threads';

if (isMainThread) {
  new Worker(new URL(import.meta.url));
  for (let n = 0; n < 1e10; n++) {
    // Looping to simulate work.
  }
} else {
  // This output will be blocked by the for loop in the main thread.
  console.log('foo');
}

const {
  Worker,
  isMainThread,
} = require('node:worker_threads');

if (isMainThread) {
  new Worker(__filename);
  for (let n = 0; n < 1e10; n++) {
    // Looping to simulate work.
  }
} else {
  // This output will be blocked by the for loop in the main thread.
  console.log('foo');
}

Запуск потоков Worker из скриптов предварительной загрузки

Будьте осторожны при запуске потоков Worker из скриптов предварительной загрузки (скриптов, загруженных и запущенных с помощью флага командной строки -r). Если параметр execArgv явно не задан, новые потоки Worker автоматически наследуют флаги командной строки запущенного процесса и предварительно загружают те же скрипты, что и основной поток. Если скрипт предварительной загрузки безусловно запускает поток Worker, каждый созданный поток будет запускать новый, пока приложение не завершится с ошибкой.