Node 工作线程池处理 CPU 密集任务

示例说明

Node 主线程是单线程，一个 35 阶斐波那契就能让所有请求卡住数十秒。worker_threads 让计算在独立线程执行，V8 实例隔离互不阻塞。线程是昂贵资源，池化复用避免频繁创建销毁；队列在全部线程忙时缓冲任务，完成回调触发补位调度。

技术要点

• 固定线程数避免线程爆炸
• busy 集合跟踪占用中的线程
• terminate 统一回收线程资源

代码解读

WorkerPool 动态把工作线程源码写入临时目录再创建 Worker，演示无独立文件时的引导方式。run 把任务包装成 Promise 入队；dispatch 取空闲线程并监听一次 message 事件，收到结果后归还线程并递归调度下一个任务。destroy 等待全部线程终止。

使用建议

生产实现建议直接使用 Piscina 这类成熟池库，它支持 transferable 与异步任务。线程间传大缓冲区用 transfer 避免拷贝。线程数默认取 CPU 核数减一，给主线程留出调度空间。

代码示例

import { Worker } from 'node:worker_threads';
import { mkdtemp, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';

const WORKER_SOURCE = `
const { parentPort, workerData } = require('node:worker_threads');
function fib(n) { return n < 2 ? n : fib(n - 1) + fib(n - 2); }
parentPort.postMessage({ input: workerData.n, result: fib(workerData.n) });
`;

class WorkerPool {
  constructor(size) {
    this.size = size;
    this.workers = [];
    this.queue = [];
    this.busy = new Set();
  }

  async init() {
    const dir = await mkdtemp(join(tmpdir(), 'pool-'));
    this.file = join(dir, 'worker.js');
    await writeFile(this.file, WORKER_SOURCE, 'utf8');
    for (let i = 0; i < this.size; i++) this.workers.push(new Worker(this.file));
  }

  run(n) {
    return new Promise((resolve, reject) => {
      this.queue.push({ n, resolve, reject });
      this.dispatch();
    });
  }

  dispatch() {
    while (this.queue.length > 0 && this.busy.size < this.size) {
      const job = this.queue.shift();
      const worker = this.workers.find((w) => !this.busy.has(w));
      this.busy.add(worker);
      worker.once('message', (msg) => {
        this.busy.delete(worker);
        job.resolve(msg.result);
        this.dispatch();
      });
      worker.postMessage({ n: job.n });
    }
  }

  async destroy() {
    await Promise.all(this.workers.map((w) => w.terminate()));
  }
}

const pool = new WorkerPool(4);
await pool.init();
console.log(await pool.run(35));
await pool.destroy();