Featured image of post Node.js 工作线程：并行处理实践Featured image of post Node.js 工作线程：并行处理实践
Web

Node.js 工作线程：并行处理实践

使用工作线程在 Node.js 中实现并行处理。涵盖线程通信、共享内存、线程池和实际用例。

  1. 首页
  2. Webs
  3. Node.js 工作线程：并行处理实践

Node.js 传统上是单线程的，依赖异步 I/O 来实现并发。虽然此模型擅长处理 I/O 密集型工作负载，但 CPU 密集型操作会阻止事件循环并降低应用程序响应能力。在 Node.js 12 中稳定的工作线程通过在单独的 V8 隔离中运行 JavaScript 来在单个进程中提供真正的并行执行。本文介绍了在生产中使用工作线程的实用模式。

工人生命周期和沟通

创建工作线程需要一个单独的 JavaScript 文件，该文件在自己的 V8 隔离中执行，并具有自己的堆和事件循环。

// main.js
const { Worker } = require("worker_threads");

const worker = new Worker("./worker.js", {
  workerData: { input: largeDataset },
});

worker.on("message", (result) => {
  logger.info({ result }, "Worker completed");
});

worker.on("error", (err) => {
  logger.error({ err }, "Worker failed");
});

worker.on("exit", (code) => {
  if (code !== 0) logger.error({ exitCode: code }, "Worker crashed");
});

// worker.js
const { parentPort, workerData } = require("worker_threads");
const result = processData(workerData.input);
parentPort.postMessage(result);

工作人员通过结构化克隆进行通信，该克隆支持对象、数组、Map、Sets、RegExp、Date 和 ArrayBuffers。对于大型二进制数据，请使用可传输对象来避免复制开销 - 源缓冲区在传输后会被中和。

使用 SharedArrayBuffer 共享内存

对于消息复制成本过高的高吞吐量场景，SharedArrayBuffer 提供线程之间的零复制共享内存。必须使用 Atomics 操作来协调访问，以防止竞争条件。

// main.js
const sharedBuffer = new SharedArrayBuffer(4 * 1024 * 1024); // 4 MB
const sharedArray = new Int32Array(sharedBuffer);
const worker = new Worker("./worker.js");
worker.postMessage({ sharedBuffer });

// Wait for worker to signal completion
Atomics.wait(sharedArray, 0, 0);
const result = sharedArray[1];

// worker.js
const { parentPort, workerData } = require("worker_threads");
const sharedArray = new Int32Array(workerData.sharedBuffer);
// Perform computation directly on shared memory
sharedArray[1] = computeResult();
Atomics.store(sharedArray, 0, 1); // Signal completion
Atomics.notify(sharedArray, 0);
机制开销使用案例
postMessage（结构化克隆）每次通话中等大多数任务，复杂的对象
可转让物品低（零拷贝）大缓冲区、二进制数据
SharedArrayBuffer + Atomics最小高频更新，流数据

线程池实现

为每个任务创建新的 Worker 实例会产生启动成本。线程池维护一组可重用的工作线程，在它们之间高效地分配任务。

class WorkerPool {
  constructor(workerPath, numThreads = os.cpus().length) {
    this.workers = [];
    this.queue = [];
    this.activeCount = 0;

    for (let i = 0; i < numThreads; i++) {
      const worker = new Worker(workerPath);
      worker.on("message", (result) => this._complete(worker, result));
      worker.on("error", (err) => this._fail(worker, err));
      this.workers.push({ worker, busy: false });
    }
  }

  execute(task) {
    return new Promise((resolve, reject) => {
      const available = this.workers.find((w) => !w.busy);
      if (available) {
        available.busy = true;
        available.worker.postMessage(task);
        available.resolve = resolve;
        available.reject = reject;
      } else {
        this.queue.push({ task, resolve, reject });
      }
    });
  }

  _complete(worker, result) {
    worker.resolve(result);
    this._next(worker);
  }

  _next(worker) {
    if (this.queue.length > 0) {
      const next = this.queue.shift();
      worker.postMessage(next.task);
      worker.resolve = next.resolve;
      worker.reject = next.reject;
    } else {
      worker.busy = false;
    }
  }
}

池大小应与 CPU 核心数相匹配。超额订阅的工作线程数多于核心数会增加上下文切换开销，而不会增加吞吐量。

用例：图像处理和数据转换

调整大小、格式转换和过滤等图像处理操作受 CPU 限制，并且在主线程上运行时会阻塞事件循环。将它们卸载到工作线程可以保持服务器的响应能力。

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

sharp(workerData.input)
  .resize(800, 600)
  .jpeg({ quality: 80 })
  .toBuffer()
  .then((output) => parentPort.postMessage(output));

工作线程还擅长处理 CPU 密集型数据任务：

  • 大负载的 JSON 解析和验证
  • CSV 和 Excel 文件处理
  • 使用 zlib 或 brotli 进行数据压缩和解压
  • 使用 bcrypt 或 argon2 进行密码散列
  • PDF 生成和渲染

基准测试通常显示，当 CPU 繁重的工作被卸载给工作线程时，p99 事件循环延迟会提高 5-10 倍，因为主线程仍然可以自由地处理传入请求。

与子进程和集群的比较

特色工作线程子进程集群
内存模型共享（同一进程）独立流程独立流程
启动时间〜5-10 毫秒~20-50 毫秒~20-50 毫秒
通讯结构化克隆+共享内存系列化IPC系列化IPC
最适合CPU 密集型任务隔离，本机插件I/O 密集型 HTTP 工作负载

cluster 模块分叉多个 Node.js 进程来处理 HTTP 请求。工作线程通过处理每个集群工作线程内的 CPU 密集型工作来补充集群：

if (cluster.isPrimary) {
  for (let i = 0; i < numCPUs; i++) cluster.fork();
} else {
  const pool = new WorkerPool("./cpu-worker.js");
  app.get("/process", async (req, res) => {
    const result = await pool.execute(req.query.data);
    res.json(result);
  });
}

监控与调试

工作线程需要特定的监视方法。监听生命周期事件，跟踪工作线程内部的内存使用情况，并使用 --inspect-brk 进行 Chrome DevTools 调试。实施运行状况检查，验证工作人员是否响应并重新启动任何已崩溃或无响应的工作人员。使用记录消息中的相关 ID 将工作程序输出与特定请求关联起来。

结论

工作线程通过为 CPU 密集型工作负载提供真正的并行执行来填补 Node.js 中的一个关键空白。通过精心设计的线程池、具有 Atomics 同步的共享内存以及仔细的任务选择，您可以显着提高应用程序吞吐量，同时保持事件循环响应。首先识别应用程序中的 CPU 密集型操作，实现具有适当错误处理的线程池，并对延迟改进进行基准测试以验证投资。

©2022 - 2026 Commentary of Takao