接下来，我们创建一个 queueItem，在其中保存传递给 .run() 方法的 getData 函数以及回调。在回调中，我们要么 resolve 或者 reject promise，这取决于 worker 是否将错误传递给回调。
如果 availableWorkerId 的值是 -1，意味着当前没有可用的 worker，我们将 queueItem 添加到 queue。如果有可用的 worker，则调用 .runWorker() 方法来执行 worker。
在 .runWorker() 方法中，我们必须把当前 worker 的 activeWorkersById 设置为使用状态；为 message 和 error 事件设置事件监听器（并在之后清理它们）；最后将数据发送给 worker。
private async runWorker(workerId: number, queueItem: QueueItem<T, N>) { const worker = this.workersById[workerId]; this.activeWorkersById[workerId] = true; const messageCallback = (result: N) => { queueItem.callback(null, result); cleanUp(); }; const errorCallback = (error: any) => { queueItem.callback(error); cleanUp(); }; const cleanUp = () => { worker.removeAllListeners('message'); worker.removeAllListeners('error'); this.activeWorkersById[workerId] = false; if (!this.queue.length) { return null; } this.runWorker(workerId, this.queue.shift()); }; worker.once('message', messageCallback); worker.once('error', errorCallback); worker.postMessage(await queueItem.getData()); }
首先，通过使用传递的 workerId，我们从 workersById 中获得 worker 引用。然后，在 activeWorkersById 中，将 [workerId] 属性设置为true，这样我们就能知道在 worker 在忙，不要运行其他任务。
接下来，分别创建 messageCallback 和 errorCallback 用来在消息和错误事件上调用，然后注册所述函数来监听事件并将数据发送给 worker。
在回调中，我们调用 queueItem 的回调，然后调用 cleanUp 函数。在 cleanUp 函数中，要删除事件侦听器，因为我们会多次重用同一个 worker。如果没有删除监听器的话就会发生内存泄漏，内存会被慢慢耗尽。
在 activeWorkersById 状态中，我们将 [workerId] 属性设置为 false，并检查队列是否为空。如果不是，就从 queue 中删除第一个项目，并用另一个 queueItem 再次调用 worker。
接着创建一个在收到 message 事件中的数据后进行一些计算的 worker：
import { isMainThread, parentPort } from 'worker_threads'; if (isMainThread) { throw new Error('Its not a worker'); } const doCalcs = (data: any) => { const collection = []; for (let i = 0; i < 1000000; i += 1) { collection[i] = Math.round(Math.random() * 100000); } return collection.sort((a, b) => { if (a > b) { return 1; } return -1; }); }; parentPort.on('message', (data: any) => { const result = doCalcs(data); parentPort.postMessage(result); });
worker 创建了一个包含 100 万个随机数的数组，然后对它们进行排序。只要能够多花费一些时间才能完成，做些什么事情并不重要。
以下是工作池简单用法的示例：
const pool = new WorkerPool<{ i: number }, number>(path.join(__dirname, './test-worker.js'), 8); const items = [...new Array(100)].fill(null); Promise.all( items.map(async (_, i) => { await pool.run(() => ({ i })); console.log('finished', i); }), ).then(() => { console.log('finished all'); });
首先创建一个由八个 worker 组成的工作池。然后创建一个包含 100 个元素的数组，对于每个元素，我们在工作池中运行一个任务。开始运行后将立即执行八个任务，其余任务被放入队列并逐个执行。通过使用工作池，我们不必每次都创建一个 worker，从而大大提高了效率。
结论
worker_threads 提供了一种为程序添加多线程支持的简单的方法。通过将繁重的 CPU 计算委托给其他线程，可以显着提高服务器的吞吐量。通过官方线程支持，我们可以期待更多来自AI、机器学习和大数据等领域的开发人员和工程师使用 Node.js.