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