本文介绍如何利用 Node.js 的 worker_threads 模块，将一个耗时的 Node.js 类自动转换为在独立的 worker 线程中运行，而无需手动编写大量的 worker 代码。通过封装一个 WrapWorker 函数，可以方便地将类及其方法暴露给主线程，实现异步执行，从而避免阻塞主线程，提高应用程序的响应能力。文章提供了详细的代码示例，并解释了其工作原理，帮助开发者快速上手。
利用 worker_threads 自动封装 Node.js 类
Node.js 的 worker_threads 模块允许你在独立的线程中执行 JavaScript 代码，这对于执行 CPU 密集型任务非常有用，可以避免阻塞主线程，提高应用程序的响应速度。手动创建和管理 worker 线程可能比较繁琐，特别是当需要暴露类的多个方法时。本文介绍一种自动将 Node.js 类转换为 worker 的方法，通过封装 WrapWorker 函数，简化了 worker 的创建和管理过程。
WrapWorker 函数详解
WrapWorker 函数接收一个类 OBJECT 和一个包含构造函数参数的数组 OBJECT_ARGS 作为输入。它在主线程中创建一个新的 worker 线程，并将类的实例在 worker 线程中实例化。然后，它通过消息传递机制，将对类方法的调用转发到 worker 线程，并将结果返回给主线程。
以下是 WrapWorker 函数的实现：
const { Worker } = require('worker_threads');
function WrapWorker(OBJECT, OBJECT_ARGS) {
if (require('worker_threads').isMainThread) {
const workerCode = `
const { parentPort, workerData } = require('worker_threads');
const OBJECT = ${OBJECT.toString()};
const instance = new OBJECT(...workerData);
parentPort.on('message', async (data) => {
try {
const method = instance[data.method];
if (typeof method === 'function') {
const response = await method.apply(instance, data.arguments);
parentPort.postMessage({ rid: data.rid, valid: true, response });
} else {
parentPort.postMessage({ rid: data.rid, valid: false });
}
} catch (error) {
parentPort.postMessage({ rid: data.rid, valid: false });
}
});
`;
const instance = new Worker(workerCode, { eval: true, workerData: OBJECT_ARGS });
const methodsProxy = {};
const methods = Object.getOwnPropertyNames(OBJECT.prototype);
let rid = 0;
const requests = {};
for (const methodId of methods) {
methodsProxy[methodId] = (...args) =>
new Promise((resolve, reject) => {
rid++;
requests[rid] = { resolve, reject };
instance.postMessage({ method: methodId, arguments: args, rid });
});
}
instance.on('message', (data) => {
const request = requests[data.rid];
if (request) {
if (data.valid) {
request.resolve(data.response);
} else {
request.reject();
}
delete requests[data.rid];
}
});
return methodsProxy;
}
}
module.exports = WrapWorker;代码解释：
- 主线程判断： if (require('worker_threads').isMainThread) 确保这段代码只在主线程中执行。
- worker 代码生成： workerCode 字符串包含了将在 worker 线程中执行的 JavaScript 代码。它将传入的 OBJECT 转换为字符串，并在 worker 线程中实例化该类。
- 消息监听： parentPort.on('message', ...) 监听来自主线程的消息。当收到消息时，它会尝试调用类的相应方法，并将结果返回给主线程。
- 错误处理： try...catch 块用于捕获 worker 线程中发生的错误，并将错误信息返回给主线程。
- worker 实例创建： const instance = new Worker(workerCode, { eval: true, workerData: OBJECT_ARGS }); 创建一个新的 worker 实例，并将 workerCode 作为 worker 的代码执行。workerData 选项用于将构造函数参数传递给 worker 线程。
- 方法代理： methodsProxy 对象用于在主线程中代理对类方法的调用。它遍历类的所有方法，并为每个方法创建一个代理函数。
- 异步 Promise 封装： 代理函数返回一个 Promise 对象，用于处理异步操作。当主线程调用代理函数时，它会将方法名和参数发送给 worker 线程，并将 Promise 对象存储在 requests 对象中。
- 消息处理： instance.on('message', ...) 监听来自 worker 线程的消息。当收到消息时，它会根据消息中的 rid 找到对应的 Promise 对象，并根据消息中的 valid 标志来 resolve 或 reject 该 Promise 对象。
使用示例
以下是如何使用 WrapWorker 函数的示例：
class RunInWorker {
constructor (starta, startb) {
this.sum = starta + startb;
}
addSync(a, b) {
return a + b + this.sum;
}
async addAsync(a, b) {
await (new Promise((resolve) => { setTimeout(resolve, 3000); }));
return a + b + this.sum;
}
}
/** @type {\RunInWorker} */
let worker = new require('./wrap-worker.js')(RunInWorker, [5, 5]);
(async () => {
console.log(worker);
console.log(await worker.addSync(3, 4), 'should be', 17);
console.log(await worker.addAsync(3, 8), 'should be', 21);
})();代码解释：
- 定义类： 首先，定义一个名为 RunInWorker 的类，该类包含一个构造函数和两个方法：addSync 和 addAsync。
- 调用 WrapWorker： 使用 require('./wrap-worker.js')(RunInWorker, [5, 5]) 调用 WrapWorker 函数，并将 RunInWorker 类和构造函数参数 [5, 5] 传递给它。这将创建一个新的 worker 线程，并在 worker 线程中实例化 RunInWorker 类。
- 方法调用： 通过 worker 对象调用 addSync 和 addAsync 方法。由于这些方法是在 worker 线程中执行的，因此它们不会阻塞主线程。
注意事项
- 确保传递给 WrapWorker 函数的类是可序列化的。
- 由于 worker 线程和主线程之间通过消息传递进行通信，因此数据的传输可能会有一定的开销。
- 在 worker 线程中发生的错误不会自动传播到主线程。需要在 worker 线程中捕获错误，并将错误信息发送给主线程。
总结
通过使用 WrapWorker 函数，可以方便地将 Node.js 类转换为 worker，从而避免阻塞主线程，提高应用程序的响应速度。这种方法简化了 worker 的创建和管理过程，并提供了一种简单的方式来暴露类的多个方法。 然而，也需要注意数据序列化和错误处理等问题，以便更好地利用 worker_threads 模块的优势。
