当前位置：首页 > news >正文

Node.js Worker Threads 实战：CPU 密集型任务的线程池设计与性能对比

news 2026/7/8 16:11:17

Node.js Worker Threads 实战：CPU 密集型任务的线程池设计与性能对比

一、事件循环的禁区：为什么加密操作能让你的 Node.js 服务卡死

Node.js 的事件循环是单线程的。当你执行JSON.parse(largeString)或crypto.scrypt()时，如果操作耗时超过 10ms，整个事件循环被阻塞——所有其他请求都在排队等待。

以下是一个真实的场景：一个图片处理 API，每次请求需要计算图片的感知哈希（Perceptual Hash）。单次计算耗时约 50ms。当并发 20 个请求同时到达时，第一个请求 50ms，第二个 100ms，第二十个 1000ms。P99 延迟从 50ms 飙升到 1s。

解决方案是把 CPU 密集型任务移到 Worker Threads 中执行。Worker Threads 是 Node.js 12+ 提供的多线程能力，每个 Worker 运行在独立的 V8 实例中，有独立的事件循环和内存空间。Worker 之间通过postMessage+ 结构化克隆传递数据。

graph TB subgraph MainThread[主线程 - Event Loop] A1[请求 1] -->|postMessage| T1[线程池] A2[请求 2] -->|postMessage| T1 A3[请求 3] -->|postMessage| T1 A1 -->|等待| W1[等待结果] B1[响应 1] -.->|on message| A1 end subgraph ThreadPool[Worker 线程池] T1 -->|分发任务| W2[Worker 1] T1 -->|分发任务| W3[Worker 2] T1 -->|分发任务| W4[Worker 3] W2 -->|CPU 密集计算| R1[结果 1] W3 -->|CPU 密集计算| R2[结果 2] W4 -->|CPU 密集计算| R3[结果 3] R1 -->|postMessage| MainThread R2 -->|postMessage| MainThread R3 -->|postMessage| MainThread end style T1 fill:#4dabf7,color:#fff style W2 fill:#51cf66,color:#fff style W3 fill:#51cf66,color:#fff style W4 fill:#51cf66,color:#fff

二、Worker Threads 的通信模型：结构化克隆的代价

Worker Threads 和主线程之间的通信基于结构化克隆算法：

数据在传递时被完整复制（不是共享内存）
支持的类型：基本类型、Array、Object、Map、Set、ArrayBuffer
不支持的类型：Function、类实例（非普通对象）、Symbol、WeakMap

这意味着每次postMessage都有序列化和反序列化的开销。对于小数据（< 1KB），这个开销可忽略。但对于大数据（如 100MB 的二进制数据），拷贝比计算本身更慢。

解决方案是使用SharedArrayBuffer——共享内存，零拷贝。Worker 直接读写主线程的同一块内存区域，配合Atomics实现同步。

对于大多数场景，结构化克隆的性能是可接受的。瓶颈通常不是数据拷贝，而是 Worker 的创建和销毁开销。

三、生产级线程池实现

// worker-pool.js const { Worker } = require('worker_threads'); const path = require('path'); class WorkerPool { constructor(workerScript, poolSize, options = {}) { this.workerScript = workerScript; this.poolSize = poolSize; this.maxQueueSize = options.maxQueueSize || 100; this.taskTimeout = options.taskTimeout || 30000; this.workers = []; this.queue = []; this.busyWorkers = new Set(); this.taskId = 0; this.pendingCallbacks = new Map(); this._init(); } _init() { for (let i = 0; i < this.poolSize; i++) { this._createWorker(i); } } _createWorker(id) { const worker = new Worker(this.workerScript, { workerData: { workerId: id }, }); worker.on('message', (result) => { this._handleResult(worker, result); }); worker.on('error', (err) => { console.error(`Worker ${id} error:`, err); this._handleWorkerFailure(worker, err); }); worker.on('exit', (code) => { if (code !== 0) { console.error(`Worker ${id} exited with code ${code}`); this._handleWorkerFailure(worker, new Error(`Worker exited with code ${code}`)); } }); worker.id = id; this.workers.push(worker); } _handleResult(worker, result) { const callback = this.pendingCallbacks.get(result.taskId); this.pendingCallbacks.delete(result.taskId); this.busyWorkers.delete(worker); // 清除超时定时器 if (callback?.timer) { clearTimeout(callback.timer); } if (result.error) { callback?.reject(new Error(result.error)); } else { callback?.resolve(result.data); } // 处理队列中的下一个任务 this._processQueue(); } _handleWorkerFailure(failedWorker, error) { // 从池中移除失败的 Worker this.workers = this.workers.filter(w => w !== failedWorker); this.busyWorkers.delete(failedWorker); // 将失败 Worker 的待处理任务重新分配 for (const [taskId, callback] of this.pendingCallbacks) { if (callback.workerId === failedWorker.id) { if (callback.timer) clearTimeout(callback.timer); callback.reject(error); this.pendingCallbacks.delete(taskId); } } // 创建新 Worker 替代 if (this.workers.length < this.poolSize) { this._createWorker(Date.now()); } } _processQueue() { if (this.queue.length === 0) return; const availableWorker = this.workers.find(w => !this.busyWorkers.has(w)); if (!availableWorker) return; const task = this.queue.shift(); this._executeTask(availableWorker, task); } _executeTask(worker, task) { const { taskId, data, resolve, reject } = task; this.busyWorkers.add(worker); const timer = setTimeout(() => { // 超时：终止 Worker 并创建新的 console.error(`Task ${taskId} timed out`); worker.terminate(); this._handleWorkerFailure(worker, new Error('Task timed out')); }, this.taskTimeout); this.pendingCallbacks.set(taskId, { resolve, reject, timer, workerId: worker.id }); worker.postMessage({ taskId, data }); } // 提交任务，返回 Promise exec(data) { if (this.queue.length >= this.maxQueueSize) { return Promise.reject(new Error('Task queue is full')); } return new Promise((resolve, reject) => { const taskId = ++this.taskId; const task = { taskId, data, resolve, reject }; const availableWorker = this.workers.find(w => !this.busyWorkers.has(w)); if (availableWorker) { this._executeTask(availableWorker, task); } else { this.queue.push(task); } }); } // 优雅关闭 async shutdown() { // 等待所有任务完成 while (this.busyWorkers.size > 0 || this.queue.length > 0) { await new Promise(resolve => setTimeout(resolve, 100)); } for (const worker of this.workers) { await worker.terminate(); } } } module.exports = { WorkerPool };

Worker 脚本（hash-worker.js）：

const { parentPort, workerData } = require('worker_threads'); const crypto = require('crypto'); console.log(`Worker ${workerData.workerId} started`); parentPort.on('message', async (msg) => { const { taskId, data } = msg; try { // CPU 密集型任务：计算 SHA-256 const hash = crypto.createHash('sha256'); if (typeof data === 'string') { hash.update(data); } else if (Buffer.isBuffer(data)) { hash.update(data); } else { hash.update(JSON.stringify(data)); } const result = hash.digest('hex'); parentPort.postMessage({ taskId, data: result }); } catch (err) { parentPort.postMessage({ taskId, error: err.message }); } });

使用示例：

const { WorkerPool } = require('./worker-pool'); const pool = new WorkerPool( path.join(__dirname, 'hash-worker.js'), os.cpus().length, // 池大小 = CPU 核心数 { maxQueueSize: 100, taskTimeout: 10000, } ); // 在 Express 路由中使用 app.post('/api/hash', async (req, res) => { try { const result = await pool.exec(req.body.text); res.json({ hash: result }); } catch (err) { res.status(500).json({ error: err.message }); } }); // 优雅关闭 process.on('SIGTERM', async () => { await pool.shutdown(); process.exit(0); });