答案：Node.js中可通过worker_threads模块创建线程池以处理CPU密集型任务，限制并发数、内存使用和任务队列长度。1. 设定最大线程数、内存（如--max-old-space-size=256）、队列长度及超时机制；2. 实现WorkerPool类管理线程生命周期与任务分配；3. worker.js执行计算任务并返回结果；4. 使用Promise控制异步任务提交；5. 建议结合piscina库增强稳定性。
在 Node.js 中，Worker Threads 适合处理 CPU 密集型任务，避免阻塞主线程。虽然 Node.js 没有内置线程池，但你可以使用 worker_threads 模块结合自定义池管理逻辑来创建一个可控的、带资源限制的 Worker 线程池。
1. 明确资源限制目标
你可能希望限制：
- 最大并发线程数：防止系统过载
- 每个线程内存使用：通过 V8 选项控制
- 任务队列长度：避免积压过多任务
- 单个任务执行超时：防止长时间卡住
2. 创建可复用的 Worker 线程池类
以下是一个简单的线程池实现，支持最大线程数限制和任务排队：
const { Worker } = require('worker_threads');
const path = require('path');
class WorkerPool {
constructor(maxWorkers = 4) {
this.maxWorkers = maxWorkers;
this.availableWorkers = [];
this.taskQueue = [];
this.workerIdCounter = 0;
// 预创建 worker
for (let i = 0; i < maxWorkers; i++) {
this.createWorker();
}}
createWorker() {
const worker = new Worker(path.resolve(__dirname, 'worker.js'));
const id = this.workerIdCounter++;
worker.on('message', (result) => {
// 执行完后释放线程并处理下一个任务
this.freeWorker(worker);
result.workerId = id;
result.success = true;
result.error = null;
if (this.taskQueue.length > 0) {
const nextTask = this.taskQueue.shift();
this.assignTask(worker, nextTask);
}
});
worker.on('error', (err) => {
this.freeWorker(worker);
if (this.taskQueue.length > 0) {
const nextTask = this.taskQueue.shift();
this.assignTask(worker, nextTask);
}
});
worker.on('exit', (code) => {
if (code !== 0) {
console.error(`Worker ${id} stopped with exit code ${code}`);
}
// 可选：重启失败的 worker
setTimeout(() => this.createWorker(), 1000);
});
this.availableWorkers.push({ worker, id });}
assignTask(worker, task) {
const index = this.availableWorkers.findIndex(w => w.worker === worker);
if (index !== -1) {
this.availableWorkers.splice(index, 1);
}
worker.postMessage(task.data);
}
freeWorker(worker) {
const exists = this.availableWorkers.some(w => w.worker === worker);
if (!exists) {
this.availableWorkers.push({ worker });
}
}
runTask(data) {
return new Promise((resolve, reject) => {
const task = { data, resolve, reject };
if (this.availableWorkers.length > 0) {
const { worker } = this.availableWorkers.pop();
this.assignTask(worker, task);
} else {
this.taskQueue.push(task);
}
});}
closeAll() {
this.availableWorkers.forEach(({ worker }) => worker.terminate());
this.taskQueue.forEach(task => {
task.reject(new Error('Pool closing, task canceled'));
});
}
}
3. 编写 Worker 执行脚本（worker.js）
这个文件是每个 Worker 实际运行的代码：
const { parentPort } = require('worker_threads');
// 模拟 CPU 密集型任务
function heavyComputation(n) {
let result = 0;
for (let i = 0; i < n; i++) {
result += Math.sqrt(i) * Math.sin(i);
}
return result;
}
parentPort.on('message', (data) => {
const result = heavyComputation(data.number || 1e7);
parentPort.postMessage({ result });
});
4. 使用线程池并设置资源边界
启动池并提交任务：
const pool = new WorkerPool(4); // 最多 4 个并发线程
async function runTasks() {
try {
const results = await Promise.all([
pool.runTask({ number: 1e7 }),
pool.runTask({ number: 1e7 }),
pool.runTask({ number: 1e7 }),
]);
console.log(results);
} catch (err) {
console.error(err);
} finally {
pool.closeAll();
}
}
runTasks();
你还可以在启动 Node.js 时限制每个 Worker 的内存：
node --max-old-space-size=256 app.js
这会将每个进程（包括 Worker）的堆内存限制为 256MB。
5. 增强功能建议
为进一步提升稳定性：
- 添加任务超时机制，使用 setTimeout 包装 Promise 并终止卡住的 Worker
- 监控 Worker 内存使用（worker.threadId + performance.memory）
- 根据 CPU 核心数动态设置 maxWorkers：require('os').cpus().length
- 使用第三方库如 piscina，它提供了更成熟的线程池和资源控制
基本上就这些。通过手动管理可用线程和任务队列，你可以精确控制并发资源，避免系统崩溃。
