Класс: Lock

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

Класс: LockManager

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

import { locks } from 'node:worker_threads';

'use strict';

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

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

import { locks } from 'node:worker_threads';

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

'use strict';

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

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

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}`);
}

'use strict';

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}`);
  }
});

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() не влияет на его состояние.

Событие: '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();

'use strict';

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, например RegExp, BigInt, Map, Set и т. д.
• value может содержать типизированные массивы, использующие как ArrayBuffer, так и SharedArrayBuffer.
• value может содержать экземпляры WebAssembly.Module.
• value не может содержать нативные объекты (на основе C++), за исключением следующих:
  • <CryptoKey>,
  • <FileHandle>,
  • <Histogram>,
  • <KeyObject>,
  • <MessagePort>,
  • <net.BlockList>,
  • <net.SocketAddress>,
  • <X509Certificate>.

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);

'use strict';

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 ]);

'use strict';

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.

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 copy

const b = Symbol('b');

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

  get d() { return 4; }
}

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

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

port2.postMessage(new Foo());

// Prints: { c: 3 } copy

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

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

port2.postMessage(new URL('https://example.org'));

// Prints: { } copy

port.hasRef()

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

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

port.ref()

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

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

port.start()

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

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

port.unref()

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

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

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, передаваемых 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; если объект нельзя клонировать (например, если он содержит function), возникает ошибка.
  • 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 и не распространяются на внешние данные, включая ArrayBuffer. Даже при заданных ограничениях процесс может аварийно завершиться при глобальной нехватке памяти.
    • 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> обещание, результатом которого является поток для чтения со снимком кучи 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.

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);
}

'use strict';

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.postMessage(value[, transferList])

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

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

worker.ref()

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

worker.resourceLimits

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

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

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

worker.startCpuProfile()

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

Запускает профилирование CPU и возвращает 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();
  const profile = await handle.stop();
  console.log(profile);
  worker.terminate();
}); copy

Пример 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();
}); copy

worker.startHeapProfile()

• Возвращает: <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();
}); copy

Пример 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();
}); copy

worker.stderr

• Тип: <stream.Readable>

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

worker.stdin

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

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

worker.stdout

• Тип: <stream.Readable>

Это поток для чтения, содержащий данные, записанные в process.stdout внутри потока Worker. Если параметр stdout: true не был передан конструктору Worker, данные передаются в поток 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.
} copy

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

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');
}

'use strict';

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, каждый созданный поток будет создавать следующий, пока приложение не завершится с ошибкой.