typescript

Node.js Worker Threads：TypeScript 并行计算实战

系统覆盖用 Worker Threads 在 Node.js/TypeScript 中做真正并行的计算实践：主线程与 worker 模型、线程安全与结构化克隆消息传递、SharedArrayBuffer 与 transferList 数据转移、Worker 池的实现、加密/图像/数据处理等 CPU 密集任务实战、与 child_process 的对比选型、资源与内存注意，以及 TypeScript 下 worker 的编译与类型边界，帮助开发者把阻塞事件循环的计算从单线程挪到多核并行。

spcent 7 分钟阅读 3249 字

引言

Node.js 是单线程事件循环：I/O 高效，但 CPU 密集任务（加密、图像处理、大 JSON、正则回溯）会卡死整个进程。worker_threads 是 Node 官方提供的真正并行方案——每个 worker 独立线程、独立 V8 实例、通过消息协作。本文讲透 worker 模型与线程安全边界、结构化克隆与 SharedArrayBuffer 两种数据转移、可复用的 Worker 池实现、三个 CPU 密集实战场景、与 child_process 的选型对比，以及资源限制与 TypeScript 编译的坑。

前置：/typescript-nodejs-backend/（Node 进程与异步）、/typescript-async-concurrency-control/（并发控制）、/typescript-error-handling-result/（跨线程错误建模）。

目录

1. 单线程模型的局限

CPU 密集任务会让事件循环「冻结」：

function hashLoop(times: number) {
  let h = 0;
  for (let i = 0; i < times; i++) h = (h * 31 + i) % 1e9;
  return h;
}

app.get("/compute", (req, res) => {
  const result = hashLoop(5_000_000); // 同步阻塞 ~500ms
  res.json({ result }); // 期间所有请求、心跳、定时器全部排队
});

阻塞后果：所有请求排队、定时器漂移（setTimeout 到点执行不了）、心跳失败被负载均衡摘除。应对手段对比：微任务拆分只缓解不根治；child_process 太重（独立进程 + 序列化）；worker_threads 同进程内并行线程——通信成本低，是 CPU 密集任务首选。

2. Worker Threads 的核心模型

核心抽象是 Worker：主线程创建，worker 跑自己的入口文件，双方用 postMessage/on 通信：

// worker.ts（worker 侧）
import { parentPort } from "node:worker_threads";

parentPort!.on("message", (msg: { times: number }) => {
  const result = hashLoop(msg.times);       // 重计算在 worker 里
  parentPort!.postMessage({ result });
});

// main.ts（主线程侧）
import { Worker } from "node:worker_threads";

const worker = new Worker(new URL("./worker.ts", import.meta.url));
worker.on("message", (m: { result: number }) => {
  console.log("结果:", m.result);
  worker.terminate();
});
worker.postMessage({ times: 5_000_000 });

关键 API：new Worker(path) 创建；worker.postMessage/parentPort.postMessage 双向发消息；on("message") 接收、on("error") 捕获 worker 未捕获异常；worker.terminate() 立即终止。模型认知：每个 worker 是独立 V8 isolate，全局互不可见；消息是唯一共享通道——没有共享内存就没有数据竞争。

3. 线程安全与消息传递

worker 间不共享 JS 对象，消息走 结构化克隆（structured clone）：

const payload = { list: [1, 2, 3], map: new Map([["a", 1]]) };
worker.postMessage(payload); // 深拷贝语义

worker.postMessage({ fn: () => {} }); // ❌ DataCloneError：函数不可克隆

线程安全的推论：无共享可变状态（主线程改 payload 不影响 worker 副本，天然规避数据竞争）；通信是异步的（不阻塞，按序到达）；worker 抛错不炸主线程，通过 error 事件传递。注意性能：大数据克隆要序列化/反序列化，此时用 §4 的转移机制。约定消息协议统一 { type, payload } 结构，方便类型判别（呼应 /typescript-typed-events-streams/）。

4. 数据转移：SharedArrayBuffer 与 transferList

大数据传输有两个优化通道。

转移所有权（零拷贝）：

const buf = new Uint8Array(1024 * 1024); // 1MB
worker.postMessage(buf.buffer, [buf.buffer]); // buffer 被转移，主线程不再可用
// 转移后 buf.buffer.byteLength === 0

共享内存（真正并行读写）：

// 主线程
const sab = new SharedArrayBuffer(1024 * 1024);
const view = new Float64Array(sab);
worker.postMessage({ sab }); // 传递引用，不复制

// worker 侧
parentPort!.on("message", (m: { sab: SharedArrayBuffer }) => {
  const view = new Float64Array(m.sab);
  Atomics.store(view, 0, 42); // 用 Atomics 保证原子性
});

铁律：共享内存 = 共享可变状态 = 数据竞争风险，必须用 Atomics（add/compareExchange/wait）同步；能用 transfer 就别用 SAB，只留给「大块数据高频交换」的场景；transfer 后所有权不可逆，主线程若还需要先复制再转移。

5. Worker 池的实现

逐个 new Worker 成本高（每 isolate 约 10-30MB 内存 + 启动耗时），生产用 Worker 池复用常驻 worker：

import { Worker } from "node:worker_threads";

type Job = { id: number; data: unknown; resolve: (v: unknown) => void };

export class WorkerPool {
  private idle: Worker[] = [];
  private queue: Job[] = [];
  private seq = 0;

  constructor(private size: number, private path: string) {
    for (let i = 0; i < size; i++) this.spawn();
  }

  private spawn() {
    const w = new Worker(this.path);
    w.on("message", (m) => { this.idle.push(w); this.fulfill(m.id, m.result); this.drain(); });
    w.on("error", () => this.spawn()); // 崩溃重建，容量恒定
    this.idle.push(w);
  }

  run<T>(data: unknown): Promise<T> {
    return new Promise((resolve, reject) => {
      const job: Job = { id: ++this.seq, data, resolve };
      const w = this.idle.pop();
      if (w) w.postMessage({ id: job.id, data: job.data });
      else this.queue.push(job);
    });
  }

  private drain() {
    while (this.queue.length && this.idle.length) {
      const job = this.queue.shift()!;
      this.idle.pop()!.postMessage({ id: job.id, data: job.data });
    }
  }
  /* fulfill / terminate 略 */
}

池参数经验：size = os.cpus().length - 1（留一核给主线程）；FIFO 分发 + 队列上限；进程退出前 terminate() 防僵尸线程。

6. CPU 密集任务实战：加密、图像与数据处理

加密（bcrypt 哈希）：

// worker.ts
import { parentPort } from "node:worker_threads";
import * as bcrypt from "bcryptjs";

parentPort!.on("message", async (m: { pw: string; rounds: number }) => {
  const hash = await bcrypt.hash(m.pw, m.rounds);
  parentPort!.postMessage({ hash });
});

图像与数据处理：worker 内 decode → resize → encode（sharp）只收发 buffer；大 JSON 转换同理——parentPort.on("message", (m: { rows }) => postMessage(m.rows.map(normalizeRow)))。

通用模式：主线程只做 I/O（读文件/收请求）与分发，计算全部进 worker。用 §5 的池封装后，调用方只面对 await pool.run(data)。

7. 与 child_process 的对比选型

child_process 是「进程级」，worker_threads 是「线程级」：

维度worker_threadschild_process（fork）
隔离线程，共享进程内存独立进程，独立内存
崩溃影响worker 崩溃不影响主进程子进程崩溃不影响父进程
数据传输结构化克隆 / transfer / SABJSON 序列化
启动成本低（同进程内）高（fork 新进程）
适合CPU 密集计算不可信代码隔离、外部工具
import { fork } from "node:child_process";
const child = fork("./script.js");
child.send({ cmd: "start" });
child.on("message", (m) => console.log(m));

选型决策：纯 JS 计算 → worker_threads；外部二进制（git/ffmpeg/python）→ child_process（exec/spawn）；要沙箱隔离不可信脚本 → child_process 或独立容器，不要 worker——worker 与主进程同崩溃域，OOM 可能拖垮整个进程。坑：fork 消息是 JSON 序列化，Buffer/TypedArray 会变形态，要显式约定二进制协议。

8. 资源与内存注意

worker 不是免费的，资源管理是生产级并行的核心：

import os from "node:os";
const MAX_WORKERS = Math.max(1, (os.availableParallelism?.() ?? os.cpus().length) - 1);

// worker 内内存自检：超阈值自毁，让池重新 spawn
parentPort!.on("message", (m) => {
  if (process.memoryUsage().heapUsed > 500 * 1024 * 1024) {
    parentPort!.postMessage({ error: "OOM" });
    process.exit(1);
  }
});

清单：内存预算按 size × 40MB 起步，别开 64 个 worker 打满小机；finally 里 terminate() 防句柄泄漏导致进程不退出；SAB 不会被 GC 回收，用完置空引用；任务产出快于消费时队列无限增长——设队列上限，满则拒绝或降级同步执行；worker 的 uncaughtException 通过 error 事件捕获并重建。坑：worker.terminate() 是异步的，频繁创建/销毁产生「锯齿内存」，池化 + 长生命周期是正解。

9. TypeScript 中的 worker 编译与类型

worker 入口在 TS 项目里要处理编译路径与类型边界：

{
  "compilerOptions": {
    "outDir": "dist", "module": "nodenext", "target": "es2022",
    "strict": true, "types": ["node"]
  }
}

TS 里写 ./worker.ts，运行期指向 dist/worker.js——用环境无关路径：

import { fileURLToPath } from "node:url";
import { dirname, join } from "node:path";
const workerPath = join(dirname(fileURLToPath(import.meta.url)), "worker.js");
new Worker(workerPath);

类型边界用判别联合定义消息协议：

type WorkerRequest =
  | { type: "hash"; times: number }
  | { type: "image"; input: Buffer; width: number };
type WorkerResponse =
  | { type: "hash"; result: number }
  | { type: "image"; output: Buffer }
  | { type: "error"; message: string };

worker.on("message", (m: WorkerResponse) => {
  if (m.type === "hash") m.result;     // ✅ number
  if (m.type === "error") m.message;   // ✅ string
});

工程要点：消息类型放共享文件（worker-protocol.ts），主线程与 worker 都 import；用 zod 校验入站消息（见 /typescript-runtime-validation-typesafe/）防版本漂移后收到未知结构。

10. 速查表与一句话记忆

场景方案
CPU 密集计算worker_threads + Worker 池
大数据传输transferList 转移所有权
高频共享数据SharedArrayBuffer + Atomics
外部二进制child_process（fork/spawn）
不可信代码隔离独立进程/容器，不用 worker
池大小os.cpus().length - 1
并发控制队列上限 + 背压
消息协议判别联合 + 运行时校验

一句话记忆：Worker Threads 并行 = 无共享消息传递（结构化克隆）+ 大块数据转移（transfer 零拷贝）/共享内存（SAB + Atomics）+ 常驻 Worker 池（容量 = 核数 - 1）+ 加密图像数据处理（计算全下放）+ 资源纪律（内存预算 + 背压 + 优雅退出）。

延伸阅读

  • /typescript-nodejs-backend/ — Node 进程模型与事件循环
  • /typescript-async-concurrency-control/ — 并发控制与背压
  • /typescript-error-handling-result/ — 跨线程错误建模
  • /typescript-typed-events-streams/ — 消息协议与判别联合
  • /typescript-runtime-validation-typesafe/ — worker 消息运行时校验
  • Node.js 专题 — child_process 与进程管理

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「typescript」更多文章

  1. TypeScript 应用安全加固：依赖、注入与敏感信息防护
  2. TypeScript Monorepo 工程化：pnpm、Turborepo 与多包协作
  3. NestJS 微服务架构：模块化、消息与网关