4、Worker Threads：CPU 密集型任务处理、线程池管理与数据共享

好，咱们来聊聊 Worker Threads。说实话，这个模块刚出来的时候，我挺兴奋的。为什么呢？因为 Node.js 单线程的痛点，终于有了一个官方的解决方案。

你想想看，Node.js 的事件循环再牛，碰上 CPU 密集型的任务——比如图像处理、数据加密、大文件压缩——也得跪。事件循环被阻塞，整个应用就像死了一样。我早期做过一个图片压缩服务，用户上传一张 4K 照片，服务器直接卡住 3 秒，其他请求全部排队。那体验，别提多糟糕了。

4.1 为什么需要 Worker Threads？

很多人会问：不是有 Cluster 吗？不是有 child_process 吗？

嗯，这里要注意。Cluster 和 child_process 确实能利用多核 CPU，但它们的问题是——太重了。每个子进程都有自己的 V8 实例，内存开销大，启动慢。而且进程间通信只能靠 IPC，数据还得序列化反序列化，效率不高。

Worker Threads 不一样。它是在同一个进程里创建新的线程。线程之间共享内存，通信成本低得多。说白了，就是轻量级的并行方案。

核心区别：

  • Cluster / child_process：多进程，独立内存，IPC 通信
  • Worker Threads：多线程，共享内存，直接数据传递

我个人习惯是：I/O 密集型用 Cluster，CPU 密集型用 Worker Threads。这个原则帮我避免了很多坑。

4.2 Worker Threads 基础用法

先看一个最简单的例子。假设我们要计算斐波那契数列的第 40 项——这玩意儿计算量不小。

// worker.js
const { parentPort } = require('worker_threads');

function fibonacci(n) {
  if (n < 2) return n;
  return fibonacci(n - 1) + fibonacci(n - 2);
}

parentPort.on('message', (n) => {
  const result = fibonacci(n);
  parentPort.postMessage(result);
});
// main.js
const { Worker } = require('worker_threads');

const worker = new Worker('./worker.js');
worker.postMessage(40);

worker.on('message', (result) => {
  console.log(`斐波那契结果: ${result}`);
  worker.terminate();
});

worker.on('error', (err) => {
  console.error('Worker 出错:', err);
});

你看，主线程把任务扔给 Worker，自己继续处理其他请求。Worker 算完了把结果发回来。主线程不会被阻塞。

我在项目中遇到过一个问题：如果 Worker 里抛了未捕获的异常，主线程是不知道的。所以一定要监听 error 事件。我曾经因为这个疏忽，导致生产环境上有个 Worker 默默挂了，数据丢了半小时才被发现。

避坑指南： 永远记得监听 Worker 的 'error' 和 'exit' 事件。Worker 崩溃时，主线程需要知道并做容错处理。

4.3 线程池管理

单个 Worker 好写，但实际项目中，你不可能来一个任务就 new 一个 Worker。创建线程是有开销的——启动时间、内存分配。频繁创建销毁，性能反而下降。

所以我们需要线程池。说白了，就是预先创建一批 Worker，任务来了就丢给空闲的 Worker 去处理。

我分享一下我自己常用的线程池实现思路：

class ThreadPool {
  constructor(size, workerPath) {
    this.size = size;
    this.workers = [];
    this.taskQueue = [];
    this.freeWorkers = [];

    for (let i = 0; i < size; i++) {
      const worker = new Worker(workerPath);
      this.workers.push(worker);
      this.freeWorkers.push(worker);

      worker.on('message', (result) => {
        // 处理完成后，把 Worker 放回空闲池
        this.freeWorkers.push(worker);
        this.processQueue();
      });
    }
  }

  runTask(data) {
    return new Promise((resolve, reject) => {
      const task = { data, resolve, reject };
      this.taskQueue.push(task);
      this.processQueue();
    });
  }

  processQueue() {
    if (this.taskQueue.length === 0) return;
    if (this.freeWorkers.length === 0) return;

    const worker = this.freeWorkers.pop();
    const task = this.taskQueue.shift();

    worker.once('message', (result) => {
      task.resolve(result);
    });
    worker.once('error', (err) => {
      task.reject(err);
    });

    worker.postMessage(task.data);
  }
}

这个实现虽然简单，但核心思想到位了。线程池大小怎么定？我一般建议等于 CPU 核心数减 1。为什么？因为主线程也要留一个核心处理 I/O 和事件循环。

CPU 核心数 推荐线程池大小 说明
2 1 留一个给主线程
4 3 适合大多数场景
8 6-7 可以适当多开
16 12-14 注意内存开销

小技巧： 可以用 os.cpus().length 动态获取 CPU 核心数，这样代码在不同机器上都能自适应。

4.4 数据共享：SharedArrayBuffer

Worker 之间传数据，默认是拷贝的。数据量大时，拷贝开销不可忽视。比如你要处理一个 100MB 的视频帧，每次 postMessage 都要拷贝一份，内存直接翻倍。

这时候 SharedArrayBuffer 就派上用场了。它允许 Worker 之间共享同一块内存，不需要拷贝。

// 主线程
const sharedBuffer = new SharedArrayBuffer(1024);
const sharedArray = new Int32Array(sharedBuffer);

const worker1 = new Worker('./worker.js');
const worker2 = new Worker('./worker.js');

worker1.postMessage({ buffer: sharedBuffer });
worker2.postMessage({ buffer: sharedBuffer });
// worker.js
const { parentPort } = require('worker_threads');

parentPort.on('message', ({ buffer }) => {
  const sharedArray = new Int32Array(buffer);
  // 直接操作共享内存
  Atomics.add(sharedArray, 0, 1);
  console.log(`当前值: ${sharedArray[0]}`);
});

这里要注意，共享内存意味着并发访问。多个 Worker 同时写同一个位置，数据就乱了。所以要用 Atomics 对象来做原子操作。

重要提醒： SharedArrayBuffer 需要设置 HTTP 响应头 'Cross-Origin-Opener-Policy: same-origin' 和 'Cross-Origin-Embedder-Policy: require-corp'。否则浏览器环境下会报错。Node.js 环境没有这个限制。

我记得有一次做实时视频处理，每帧数据 2MB，如果用拷贝方式，内存占用飙升到 8GB。改用 SharedArrayBuffer 后，内存降到 2GB，性能提升 3 倍。这就是共享内存的威力。

4.5 实战：图片缩略图生成服务

咱们把前面学的串起来，做一个图片缩略图生成服务。这个需求很常见——用户上传图片，服务器生成多种尺寸的缩略图。

// thumbnail-worker.js
const { parentPort, workerData } = require('worker_threads');
const sharp = require('sharp');

async function generateThumbnail(inputPath, outputPath, width, height) {
  await sharp(inputPath)
    .resize(width, height)
    .toFile(outputPath);
  return { outputPath, width, height };
}

parentPort.on('message', async (task) => {
  try {
    const result = await generateThumbnail(
      task.inputPath,
      task.outputPath,
      task.width,
      task.height
    );
    parentPort.postMessage({ success: true, data: result });
  } catch (err) {
    parentPort.postMessage({ success: false, error: err.message });
  }
});
// 主服务
const { Worker } = require('worker_threads');
const path = require('path');

const pool = new ThreadPool(4, './thumbnail-worker.js');

app.post('/upload', async (req, res) => {
  const inputPath = req.file.path;
  const sizes = [
    { width: 150, height: 150 },
    { width: 300, height: 300 },
    { width: 600, height: 600 }
  ];

  try {
    const results = await Promise.all(
      sizes.map(size => pool.runTask({
        inputPath,
        outputPath: `thumb_${size.width}_${path.basename(inputPath)}`,
        ...size
      }))
    );
    res.json({ success: true, thumbs: results });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

这个例子中，线程池管理着 4 个 Worker。每个上传请求需要生成 3 种尺寸的缩略图，正好可以并行处理。主线程只负责接收请求和返回结果，CPU 密集的图像处理全交给 Worker。

性能对比： 单线程处理 100 张图片需要 45 秒。用 4 个 Worker 的线程池，只需要 12 秒。接近线性加速。

4.6 常见陷阱与最佳实践

最后，分享几个我踩过的坑：

  • Worker 数量不是越多越好。 线程切换有开销，超过 CPU 核心数反而会降低性能。
  • 避免在 Worker 里操作 DOM。 虽然 Node.js 没有 DOM，但有些库会尝试访问，导致崩溃。
  • Worker 里的 console.log 不会显示在主进程。 需要手动通过 parentPort 传递日志。
  • 不要共享复杂对象。 SharedArrayBuffer 只适合二进制数据。复杂对象还是用 postMessage 拷贝吧。

我曾经在生产环境上犯过一个错误：线程池大小设成了 16，而服务器只有 4 核。结果 CPU 上下文切换开销巨大，性能反而比单线程还差。后来改成 3 个 Worker，一切恢复正常。

嗯，Worker Threads 的内容就这些。记住核心思路：CPU 密集型任务交给 Worker，主线程保持轻量。线程池管理好，数据共享用 SharedArrayBuffer。这样你的 Node.js 应用才能真正扛住高并发。

← 上一章 📖 返回目录 下一章 →