目录

• worker_threads（工作线程）
  • worker.isMainThread
  • worker.markAsUntransferable(object)
  • worker.moveMessagePortToContext(port, contextifiedSandbox)
  • worker.parentPort
  • worker.receiveMessageOnPort(port)
  • worker.resourceLimits
  • worker.SHARE_ENV
  • worker.threadId
  • worker.workerData
  • MessageChannel 类
  • MessagePort 类
    • 'close' 事件
    • 'message' 事件
    • 'messageerror' 事件
    • port.close()
    • port.postMessage(value[, transferList])
      • 传输 TypedArray 和 Buffer 时的注意事项
    • port.ref()
    • port.start()
    • port.unref()
  • Worker 类
    • new Worker(filename[, options])
    • 'error' 事件
    • 'exit' 事件
    • 'message' 事件
    • 'messageerror' 事件
    • 'online' 事件
    • worker.getHeapSnapshot()
    • worker.postMessage(value[, transferList])
    • worker.ref()
    • worker.resourceLimits
    • worker.stderr
    • worker.stdin
    • worker.stdout
    • worker.terminate()
    • worker.threadId
    • worker.unref()

worker_threads（工作线程）#

源代码: lib/worker_threads.js

worker_threads 模块允许使用并行地执行 JavaScript 的线程。 要访问它：

const worker = require('worker_threads');

工作线程对于执行 CPU 密集型的 JavaScript 操作非常有用。 它们在 I/O 密集型的工作中用途不大。 Node.js 的内置的异步 I/O 操作比工作线程效率更高。

与 child_process 或 cluster 不同， worker_threads 可以共享内存。 它们通过传输 ArrayBuffer 实例或共享 SharedArrayBuffer 实例来实现。

const {
  Worker, isMainThread, parentPort, workerData
} = require('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.on('error', reject);
      worker.on('exit', (code) => {
        if (code !== 0)
          reject(new Error(`工作线程使用退出码 ${code} 停止`));
      });
    });
  };
} else {
  const { parse } = require('一些 js 解析库');
  const script = workerData;
  parentPort.postMessage(parse(script));
}

上面的示例为每个 parse() 调用衍生一个工作线程。 在实际的实践中，应使用工作线程池代替这些任务。 否则，创建工作线程的开销可能会超出其收益。

当实现工作线程池时，可使用 AsyncResource API 来通知诊断的工具（例如为了提供异步的堆栈跟踪）有关任务及其结果之间的相关性。 有关示例的实现，请参见 async_hooks 文档中的“为 Worker 线程池使用 AsyncResource”。

默认情况下，工作线程继承非特定于进程的选项。 请参见工作线程的构造函数选项，以了解如何自定义工作线程的选项，特别是 argv 和 execArgv 选项。

worker.isMainThread#

• <boolean>

如果此代码不在 Worker 线程内运行，则为 true。

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

if (isMainThread) {
  // 这会在工作线程实例中重新加载当前文件。
  new Worker(__filename);
} else {
  console.log('在工作线程中');
  console.log(isMainThread);  // 打印 'false'。
}

worker.markAsUntransferable(object)#

Mark an object as not transferable. If object occurs in the transfer list of a port.postMessage() call, it will be ignored.

In particular, this makes sense for objects that can be cloned, rather than transferred, and which are used by other objects on the sending side. For example, Node.js marks the ArrayBuffers it uses for its Buffer pool with this.

This operation cannot be undone.

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

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

markAsUntransferable(pooledBuffer);

const { port1 } = new MessageChannel();
port1.postMessage(typedArray1, [ typedArray1.buffer ]);

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

There is no equivalent to this API in browsers.

worker.moveMessagePortToContext(port, contextifiedSandbox)#

• port <MessagePort> The message port which will be transferred.

• contextifiedSandbox <Object> A contextified object as returned by the vm.createContext() method.

• Returns: <MessagePort>

Transfer a MessagePort to a different vm Context. The original port object will be rendered unusable, and the returned MessagePort instance will take its place.

The returned MessagePort will be an object in the target context, and will inherit from its global Object class. Objects passed to the port.onmessage() listener will also be created in the target context and inherit from its global Object class.

However, the created MessagePort will no longer inherit from EventEmitter, and only port.onmessage() can be used to receive events using it.

worker.parentPort#

• <null> | <MessagePort>

如果当前线程为Worker工作线程, 该MessagePort端口作用于与主线交换信息。通过该端口parentPort.postMessage()发送的消息在主线程中将可以通过worker.on('message')接收。主线程中通过worker.postMessage()发送的消息将可以在工作线程中通过parentPort.on('message')接收。

const { Worker, isMainThread, parentPort } = require('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.receiveMessageOnPort(port)#

• port <MessagePort>

• Returns: <Object> | <undefined>

Receive a single message from a given MessagePort. If no message is available, undefined is returned, otherwise an object with a single message property that contains the message payload, corresponding to the oldest message in the MessagePort’s queue.

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

When this function is used, no 'message' event will be emitted and the onmessage listener will not be invoked.

worker.resourceLimits#

• <Object>
  • maxYoungGenerationSizeMb <number>
  • maxOldGenerationSizeMb <number>
  • codeRangeSizeMb <number>
  • stackSizeMb <number>

Provides the set of JS engine resource constraints inside this Worker thread. If the resourceLimits option was passed to the Worker constructor, this matches its values.

If this is used in the main thread, its value is an empty object.

worker.SHARE_ENV#

• <symbol>

传递给构造函数Worker选项对象env属性的值，用以指定主线程与工作线程将可共享环境变量的读写。

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

worker.threadId#

• <integer>

当前线程的ID，同时在工作线程上，每个worker实例也都有唯一的ID。

worker.workerData#

（工作线程中可用）指代通过主线程中传递过来的数据。 它可以是任意的JavaScript值，通过主线程构造函数中的选项对象的workerData传递。 这个数据类似Web Worker中postMessage()机制，它是拷贝传递的（所以如果是较大数据里，不建议通过此方法）。

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

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

MessageChannel 类#

Instances of the worker.MessageChannel class represent an asynchronous, two-way communications channel. The MessageChannel has no methods of its own. new MessageChannel() yields an object with port1 and port2 properties, which refer to linked MessagePort instances.

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>

Instances of the worker.MessagePort class represent one end of an asynchronous, two-way communications channel. It can be used to transfer structured data, memory regions and other MessagePorts between different Workers.

With the exception of MessagePorts being EventEmitters rather than EventTargets, this implementation matches browser MessagePorts.

'close' 事件#

The 'close' event is emitted once either side of the channel has been disconnected.

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

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

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

'message' 事件#

• value <any> The transmitted value

The 'message' event is emitted for any incoming message, containing the cloned input of port.postMessage().

Listeners on this event will receive a clone of the value parameter as passed to postMessage() and no further arguments.

'messageerror' 事件#

• error <Error> An Error object

The 'messageerror' event is emitted when deserializing a message failed.

port.close()#

Disables further sending of messages on either side of the connection. This method can be called when no further communication will happen over this MessagePort.

The 'close' event will be emitted on both MessagePort instances that are part of the channel.

port.postMessage(value[, transferList])#

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

Sends a JavaScript value to the receiving side of this channel. value will be transferred in a way which is compatible with the HTML structured clone algorithm.

In particular, the significant differences to JSON are:

• value may contain circular references.
• value may contain instances of builtin JS types such as RegExps, BigInts, Maps, Sets, etc.
• value may contain typed arrays, both using ArrayBuffers and SharedArrayBuffers.
• value may contain WebAssembly.Module instances.
• value may not contain native (C++-backed) objects other than MessagePorts, FileHandles, and KeyObjects.

const { MessageChannel } = require('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 may be a list of ArrayBuffer, MessagePort and FileHandle objects. After transferring, they will not be usable on the sending side of the channel anymore (even if they are not contained in value). Unlike with child processes, transferring handles such as network sockets is currently not supported.

If value contains SharedArrayBuffer instances, those will be accessible from either thread. They cannot be listed in transferList.

value may still contain ArrayBuffer instances that are not in transferList; in that case, the underlying memory is copied rather than moved.

const { MessageChannel } = require('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` will be 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 ]);

Because the object cloning uses the structured clone algorithm, non-enumerable properties, property accessors, and object prototypes are not preserved. In particular, Buffer objects will be read as plain Uint8Arrays on the receiving side.

The message object will be cloned immediately, and can be modified after posting without having side effects.

For more information on the serialization and deserialization mechanisms behind this API, see the serialization API of the v8 module.

传输 TypedArray 和 Buffer 时的注意事项#

All TypedArray and Buffer instances are views over an underlying ArrayBuffer. That is, it is the ArrayBuffer that actually stores the raw data while the TypedArray and Buffer objects provide a way of viewing and manipulating the data. It is possible and common for multiple views to be created over the same ArrayBuffer instance. Great care must be taken when using a transfer list to transfer an ArrayBuffer as doing so will cause all TypedArray and Buffer instances that share that same ArrayBuffer to become unusable.

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

For Buffer instances, specifically, whether the underlying ArrayBuffer can be transferred or cloned depends entirely on how instances were created, which often cannot be reliably determined.

An ArrayBuffer can be marked with markAsUntransferable() to indicate that it should always be cloned and never transferred.

Depending on how a Buffer instance was created, it may or may not own its underlying ArrayBuffer. An ArrayBuffer must not be transferred unless it is known that the Buffer instance owns it. In particular, for Buffers created from the internal Buffer pool (using, for instance Buffer.from() or Buffer.alloc()), transferring them is not possible and they will always be cloned, which sends a copy of the entire Buffer pool. This behavior may come with unintended higher memory usage and possible security concerns.

See Buffer.allocUnsafe() for more details on Buffer pooling.

The ArrayBuffers for Buffer instances created using Buffer.alloc() or Buffer.allocUnsafeSlow() can always be transferred but doing so will render all other existing views of those ArrayBuffers unusable.

port.ref()#

Opposite of unref(). Calling ref() on a previously unref()ed port will not let the program exit if it's the only active handle left (the default behavior). If the port is ref()ed, calling ref() again will have no effect.

If listeners are attached or removed using .on('message'), the port will be ref()ed and unref()ed automatically depending on whether listeners for the event exist.

port.start()#

Starts receiving messages on this MessagePort. When using this port as an event emitter, this will be called automatically once 'message' listeners are attached.

This method exists for parity with the Web MessagePort API. In Node.js, it is only useful for ignoring messages when no event listener is present. Node.js also diverges in its handling of .onmessage. Setting it will automatically call .start(), but unsetting it will let messages queue up until a new handler is set or the port is discarded.

port.unref()#

Calling unref() on a port will allow the thread to exit if this is the only active handle in the event system. If the port is already unref()ed calling unref() again will have no effect.

If listeners are attached or removed using .on('message'), the port will be ref()ed and unref()ed automatically depending on whether listeners for the event exist.

Worker 类#

• 继承自: <EventEmitter>

Worker 类代表一个独立的 JavaScript 执行线程。 大多数 Node.js API 都在其中可用。

工作线程环境中的显着差异是：

• 父线程可以重定向 process.stdin、process.stdout 和 process.stderr。
• require('worker_threads').isMainThread 属性被设置为 false。
• require('worker_threads').parentPort 消息端口可用。
• process.exit() 不会停止整个程序，仅停止单个线程，且 process.abort() 不可用。
• process.chdir() 和设置群组或用户标识的 process 方法不可用。
• process.env 是父线程的环境变量的副本，除非另外指定。 对一个副本的更改将会在其他线程中不可见，并且对原生插件也不可见（除非 worker.SHARE_ENV 作为 env 选项传给 Worker 的构造函数）。
• process.title 无法被修改。
• 信号将不会通过 process.on('...') 传递。
• 调用 worker.terminate() 可能会随时停止执行。
• 无法访问父进程的 IPC 通道。
• 不支持 trace_events 模块。
• 如果原生插件满足特定条件，则只能从多个线程中加载它们。

可以在其他 Worker 实例中创建 Worker 实例。

与 Web 工作线程和 cluster 模块一样，可以通过线程间的消息传递来实现双向通信。 在内部，一个 Worker 具有一对内置的 MessagePort，在创建该 Worker 时它们已经相互关联。 虽然父端的 MessagePort 对象没有直接公开，但其功能是通过父线程的 Worker 对象上的 worker.postMessage() 和 worker.on('message') 事件公开的。

要创建自定义的消息传递通道（建议使用默认的全局通道，因为这样可以促进关联点的分离），用户可以在任一线程上创建一个 MessageChannel 对象，并将该 MessageChannel 上的 MessagePort 中的一个通过预先存在的通道传给另一个线程，例如全局的通道。

有关如何传递消息以及可以通过线程屏障成功地传输哪类 JavaScript 值的更多信息，请参见 port.postMessage()。

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('接收到:', value);
  });
} else {
  parentPort.once('message', (value) => {
    assert(value.hereIsYourPort instanceof MessagePort);
    value.hereIsYourPort.postMessage('工作线程正在发送此消息');
    value.hereIsYourPort.close();
  });
}

new Worker(filename[, options])#

• filename <string> 工作线程主脚本的路径。必须是以 ./ 或 ../ 开头的绝对路径或相对路径（即相对于当前工作目录）、或者使用 file: 或 data: 协议的 WHATWG URL 对象。 When using a data: URL, the data is interpreted based on MIME type using the ECMAScript module loader. 如果 options.eval 为 true，则这是一个包含 JavaScript 代码而不是路径的字符串。
• options <Object>
  • argv <any[]> 参数列表，其将会被字符串化并附加到工作线程中的 process.argv。 这大部分与 workerData 相似，但是这些值将会在全局的 process.argv 中可用，就好像它们是作为 CLI 选项传给脚本一样。
  • env <Object> 如果设置，则指定工作线程中 process.env 的初始值。 作为一个特殊值，worker.SHARE_ENV 可以用于指定父线程和子线程应该共享它们的环境变量。 在这种情况下，对一个线程的 process.env 对象的更改也会影响另一个线程。 默认值: process.env。
  • eval <boolean> 如果为 true 且第一个参数是一个 string，则将构造函数的第一个参数解释为工作线程联机后执行的脚本。
  • execArgv <string[]> 传递给工作线程的 node CLI 选项的列表。 不支持 V8 选项（例如 --max-old-space-size）和影响进程的选项（例如 --title）。 如果设置，则它将会作为工作线程内部的 process.execArgv 提供。 默认情况下，选项将会从父线程继承。
  • stdin <boolean> 如果将其设置为 true，则 worker.stdin 将会提供一个可写流，其内容将会在工作线程中以 process.stdin 出现。 默认情况下，不提供任何数据。
  • stdout <boolean> 如果将其设置为 true，则 worker.stdout 将不会自动地通过管道传递到父线程中的 process.stdout。
  • stderr <boolean> 如果将其设置为 true，则 worker.stderr 将不会自动地通过管道传递到父线程中的 process.stderr。
  • workerData <any> 能被克隆并作为 require('worker_threads').workerData 的任何 JavaScript 值。 克隆将会按照 HTML 结构化克隆算法中描述的进行，如果对象无法被克隆（例如，因为它包含 function），则会抛出错误。
  • trackUnmanagedFds <boolean> If this is set to true, then the Worker will track raw file descriptors managed through fs.open() and fs.close(), and close them when the Worker exits, similar to other resources like network sockets or file descriptors managed through the FileHandle API. This option is automatically inherited by all nested Workers. Default: false.
  • transferList <Object[]> If one or more MessagePort-like objects are passed in workerData, a transferList is required for those items or ERR_MISSING_MESSAGE_PORT_IN_TRANSFER_LIST will be thrown. See port.postMessage() for more information.
  • resourceLimits <Object> 新的 JS 引擎实例的一组可选的资源限制。 达到这些限制将会导致终止 Worker 实例。 这些限制仅影响 JS 引擎，并且不影响任何外部数据，包括 ArrayBuffers。 即使设置了这些限制，如果遇到全局内存不足的情况，该进程仍可能中止。
    • maxOldGenerationSizeMb <number> 主堆的最大大小，以 MB 为单位。
    • maxYoungGenerationSizeMb <number> 最近创建的对象的堆空间的最大大小。
    • codeRangeSizeMb <number> 用于生成代码的预分配的内存范围的大小。
    • stackSizeMb <number> The default maximum stack size for the thread. Small values may lead to unusable Worker instances. Default: 4.

'error' 事件#

• err <Error>

The 'error' event is emitted if the worker thread throws an uncaught exception. In that case, the worker will be terminated.

'exit' 事件#

• exitCode <integer>

The 'exit' event is emitted once the worker has stopped. If the worker exited by calling process.exit(), the exitCode parameter will be the passed exit code. If the worker was terminated, the exitCode parameter will be 1.

This is the final event emitted by any Worker instance.

'message' 事件#

• value <any> The transmitted value

The 'message' event is emitted when the worker thread has invoked require('worker_threads').parentPort.postMessage(). See the port.on('message') event for more details.

All messages sent from the worker thread will be emitted before the 'exit' event is emitted on the Worker object.

'messageerror' 事件#

• error <Error> An Error object

The 'messageerror' event is emitted when deserializing a message failed.

'online' 事件#

The 'online' event is emitted when the worker thread has started executing JavaScript code.

worker.getHeapSnapshot()#

• Returns: <Promise> A promise for a Readable Stream containing a V8 heap snapshot

Returns a readable stream for a V8 snapshot of the current state of the Worker. See v8.getHeapSnapshot() for more details.

If the Worker thread is no longer running, which may occur before the 'exit' event is emitted, the returned Promise will be rejected immediately with an ERR_WORKER_NOT_RUNNING error.

worker.postMessage(value[, transferList])#

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

Send a message to the worker that will be received via require('worker_threads').parentPort.on('message'). See port.postMessage() for more details.

worker.ref()#

Opposite of unref(), calling ref() on a previously unref()ed worker will not let the program exit if it's the only active handle left (the default behavior). If the worker is ref()ed, calling ref() again will have no effect.

worker.resourceLimits#

• <Object>
  • maxYoungGenerationSizeMb <number>
  • maxOldGenerationSizeMb <number>
  • codeRangeSizeMb <number>
  • stackSizeMb <number>

Provides the set of JS engine resource constraints for this Worker thread. If the resourceLimits option was passed to the Worker constructor, this matches its values.

If the worker has stopped, the return value is an empty object.

worker.stderr#

• <stream.Readable>

This is a readable stream which contains data written to process.stderr inside the worker thread. If stderr: true was not passed to the Worker constructor, then data will be piped to the parent thread's process.stderr stream.

worker.stdin#

• <null> | <stream.Writable>

If stdin: true was passed to the Worker constructor, this is a writable stream. The data written to this stream will be made available in the worker thread as process.stdin.

worker.stdout#

• <stream.Readable>

This is a readable stream which contains data written to process.stdout inside the worker thread. If stdout: true was not passed to the Worker constructor, then data will be piped to the parent thread's process.stdout stream.

worker.terminate()#

• Returns: <Promise>

Stop all JavaScript execution in the worker thread as soon as possible. Returns a Promise for the exit code that is fulfilled when the 'exit' event is emitted.

worker.threadId#

• <integer>

An integer identifier for the referenced thread. Inside the worker thread, it is available as require('worker_threads').threadId. This value is unique for each Worker instance inside a single process.

worker.unref()#

Calling unref() on a worker will allow the thread to exit if this is the only active handle in the event system. If the worker is already unref()ed calling unref() again will have no effect.