Потоки обработки

Модуль worker предоставляет способ создания нескольких сред, работающих в независимых потоках, и создания каналов обмена сообщениями между ними. К нему можно получить доступ с помощью флага --experimental-worker и:

const worker = require('worker_threads');

Потоки полезны для выполнения ресурсоёмких операций JavaScript; не используйте их для ввода-вывода, так как встроенные механизмы Node.js для асинхронных операций уже обрабатывают их более эффективно, чем потоки обработки.

Потоки, в отличие от дочерних процессов или при использовании модуля cluster, также могут эффективно обмениваться памятью, передавая экземпляры ArrayBuffer или совмещая экземпляры SharedArrayBuffer между ними.

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

if (isMainThread) {
  module.exports = async function parseJSAsync(script) {
    return new Promise((resolve, reject) => {
      const worker = new Worker(__filename, {
        workerData: script
      });
      worker.on('message', resolve);
      worker.on('error', reject);
      worker.on('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));
}

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

worker.isMainThread

• <boolean>

Возвращает true, если данный код не выполняется внутри потока Worker.

worker.parentPort

• <null> | <MessagePort>

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

worker.threadId

• <целое>

Целое число, идентификатор текущего потока. В соответствующем объекте потока (если он есть), он доступен как worker.threadId.

worker.workerData

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

Класс: MessageChannel

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

const { MessageChannel } = require('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

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

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

За исключением того, что MessagePort являются объектами EventEmitter, а не EventTarget, эта реализация соответствует браузерным MessagePort.

Событие: 'close'

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

Событие: 'message'

• value <любой> Переданное значение

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

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

port.close()

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

port.postMessage(value[, transferList])

• value <любой>
• transferList <Массив объектов>

Отправляет значение JavaScript на принимающую сторону этого канала. value будет передаваться таким образом, что совместимо с алгоритмом структурированного клонирования HTML. В частности, он может содержать циклические ссылки и объекты, такие как массивы с типом данных, которые API JSON не может сериализовать.

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

Если value содержит экземпляры SharedArrayBuffer, они будут доступны в любом потоке. Они не могут быть перечислены в transferList.

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

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

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

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

port.ref()

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

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

port.start()

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

port.unref()

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

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

Класс: Worker

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

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

Существенные различия в среде Worker:

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

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

• Модуль inspector пока недоступен.
• Нативные плагины пока не поддерживаются.

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

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

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

См. port.postMessage() для получения дополнительной информации о передаче сообщений и о типах JavaScript-значений, которые могут быть успешно перенесены через потоковую границу.

const assert = require('assert');
const {
  Worker, MessageChannel, MessagePort, isMainThread, parentPort
} = require('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 <строка> Путь к основному скрипту Worker. Должен быть либо абсолютным, либо относительным путём (относительно текущей рабочей директории), начинающимся с ./ или ../. Если options.eval равно true, это строка с кодом JavaScript, а не путь.

• options <Объект>

  • eval <логическое значение> Если true, интерпретировать первый аргумент конструктора как скрипт, который выполняется, как только рабочий процесс готов.
  • workerData <любое> Любое JavaScript-значение, которое будет клонировано и доступно как require('worker_threads').workerData. Клонирование будет происходить в соответствии с алгоритмом структурированного клонирования HTML, и будет выброшено исключение, если объект не может быть клонирован (например, из-за наличия function).
  • stdin <логическое значение> Если установлено значение true, то worker.stdin предоставит потоковый объект для записи, содержимое которого будет отображаться как process.stdin внутри Worker. По умолчанию данные не предоставляются.
  • stdout <логическое значение> Если установлено значение true, то worker.stdout не будет автоматически перенаправлен в process.stdout родителя.
  • stderr <логическое значение> Если установлено значение true, то worker.stderr не будет автоматически перенаправлен в process.stderr родителя.

Событие: 'error'

• err <Ошибка>

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

Событие: 'exit'

• exitCode <целое число>

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

Событие: 'message'

• value <любое> Переданное значение

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

Событие: 'online'

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

worker.postMessage(value[, transferList])

• value <любое>
• transferList <массив объектов>

Отправить сообщение рабочему процессу, которое будет получено через require('worker_threads').parentPort.on('message'). См. port.postMessage() для получения дополнительной информации.

worker.ref()

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

worker.stderr

• <stream.Readable>

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

worker.stdin

• <null> | <stream.Writable>

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

worker.stdout

• <stream.Readable>

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

worker.terminate([callback])

• callback <Функция>

  • err <Ошибка>
  • exitCode <целое число>

Прекратить выполнение всего JavaScript-кода в рабочем потоке как можно скорее. callback — это необязательная функция, которая вызывается, как только эта операция будет завершена.

Предупреждение: В настоящее время не весь код в ядре Node.js подготовлен к ожиданию завершения в произвольные моменты времени и может аварийно завершиться при столкновении с такой ситуацией. Следовательно, вызывайте .terminate() только в том случае, если известно, что поток Worker не обращается к ядру Node.js, за исключением модулей, доступных в модуле worker.

worker.threadId

• <целое>

Целое число, идентификатор связанного потока. Внутри потока Worker он доступен как require('worker_threads').threadId.

worker.unref()

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