背景与问题

很多人第一次用 Node.js 做任务处理服务时，都会先走一条“看起来很顺”的路：

• HTTP 接口收请求
• 直接在主线程里算
• 算完返回结果
• 并发一高，CPU 飙升，接口开始超时

这条路在 I/O 型业务里通常没问题，但一旦任务带有明显 CPU 消耗，比如：

• 批量图片处理
• 文本分词、加密、压缩
• 大 JSON 解析与转换
• 风控规则批量计算
• 数据报表聚合

主线程就会被堵住。Node.js 的事件循环一旦被长时间占用，最先受伤的不是“计算任务本身”，而是整台服务的响应能力：健康检查变慢、接口抖动、超时增多，甚至日志都开始延迟输出。

我自己踩过一个很典型的坑：业务方以为“只是加了一个导出报表接口”，结果这个接口在高峰期把整个 API 服务拖慢。后来排查发现，不是数据库慢，而是主线程被 JSON 序列化和聚合逻辑卡住了。

所以这篇文章不只是讲“怎么用 Worker Threads”，而是从架构角度把一个可落地的方案串起来：

1. CPU 密集任务放进 Worker Threads
2. 主线程只负责调度、接入与健康保护
3. 通过事件循环监控判断系统是否过载
4. 在高并发下做限流、降级、超时与回收

这比单纯“开几个 worker”更接近真实生产环境。

先看整体架构

我们先建立一个清晰的心智模型：主线程不做重活，只做交通警察。

flowchart TD
    A[客户端请求] --> B[Node.js 主线程 HTTP 服务]
    B --> C[任务队列]
    B --> D[事件循环监控]
    C --> E[Worker Pool 调度器]
    E --> F[Worker 1]
    E --> G[Worker 2]
    E --> H[Worker N]
    F --> I[任务结果]
    G --> I
    H --> I
    D --> J{系统过载?}
    J -- 否 --> C
    J -- 是 --> K[拒绝/降级/限流]
    I --> B
    B --> L[响应客户端]

这个架构有几个关键点：

• 主线程：负责接收请求、排队、调度、监控
• Worker Pool：固定数量线程池，而不是每个任务新开一个线程
• 事件循环监控：检测主线程是否已经堵住
• 过载保护：不能无限接任务，否则只是把问题从“阻塞”变成“雪崩”

核心原理

1. 为什么单线程 Node.js 会在 CPU 任务下失速

Node.js 的强项是事件驱动和异步 I/O。但 JavaScript 执行本身仍然依赖主线程事件循环。只要某段同步计算执行太久，事件循环就没机会处理：

• 新的 HTTP 请求
• 定时器回调
• Promise 后续任务
• socket 事件
• 健康检查

这也是为什么“CPU 密集型任务不适合直接跑在主线程”。

2. Worker Threads 解决了什么问题

worker_threads 是 Node.js 提供的真正线程能力。它适合把 CPU 密集型计算 挪到独立线程中执行。

注意两个边界：

• 适合：计算、解析、压缩、加密、复杂转换
• 不适合：拿它替代普通异步 I/O，比如数据库查询、HTTP 请求；这些仍应优先走 Node 的异步模型

Worker 与主线程之间通过消息通信。数据会经历结构化拷贝，必要时可用 Transferable 或 SharedArrayBuffer 优化。

3. 为什么要用线程池，而不是每次 new Worker

如果你每来一个任务就 new Worker()：

• 线程创建有成本
• 内存占用不可控
• 高频任务下线程爆炸
• GC 和上下文切换会拖慢整体吞吐

更合理的做法是：

• 预创建固定数量 Worker
• 主线程维护待处理队列
• 空闲 Worker 领取任务
• 任务完成后回到池中

这就是经典的 Worker Pool。

4. 事件循环监控的意义

很多人会说：“既然计算都挪进 Worker 了，还监控事件循环干嘛？”

因为主线程仍然可能被压垮，原因包括：

• 请求太多，排队逻辑本身变重
• 参数校验过于复杂
• 日志序列化、响应封装耗 CPU
• 大对象消息传递带来拷贝成本
• 内存压力导致 GC 停顿

Node.js 提供了 perf_hooks.monitorEventLoopDelay()，可以帮助我们观测事件循环延迟。这个指标比单看 CPU 更贴近“服务是否还灵”。

常见理解方式：

• 延迟小：主线程比较轻松
• 延迟持续升高：主线程在卡顿，服务开始失去实时性

5. 架构取舍：Worker Threads vs Cluster vs 外部队列

这是中级实战里很重要的一步：不要一上来就“用最新 API”，而是先知道边界。

这篇文章的定位是：单机或单服务内部，构建高并发 CPU 任务处理能力。如果你的任务已经跨机器、需要持久化重试、需要多级优先级，那就该考虑 MQ 了。

方案设计与容量估算

在架构设计上，我建议先做一个“够用且稳定”的版本，而不是一开始追求极致吞吐。

线程池大小怎么定

经验上可以先从：

worker 数量 = CPU 核数 - 1

开始试。

为什么不是越多越好？

• 线程多了会有上下文切换开销
• 任务若本身占用大量内存，会把 RSS 顶上去
• 主线程还需要资源处理接入和调度

如果你的任务比较“重”，甚至可以从 CPU 核数 / 2 开始测。

队列长度怎么定

队列不是越长越安全，太长意味着：

• 请求响应时间不可控
• 内存堆积
• 用户拿到结果太晚，业务上已经失去意义

一个实用思路：

最大排队长度 = worker 数量 * 每线程可接受排队倍数

比如：

• 8 个 worker
• 每个 worker 最多接受 20 个待处理任务

那总队列上限可先设为 160。超过直接拒绝或降级，而不是“先收着再说”。

超时要分两层

• 排队超时：任务在队列里等太久，说明系统已忙
• 执行超时：Worker 真正执行超时，可能是数据异常或逻辑退化

这两种超时不能混在一起，否则排查会很痛苦。

实战代码（可运行）

下面我们实现一个简化但可运行的版本：

• 一个 HTTP 服务
• 一个固定大小 Worker Pool
• 一个事件循环监控器
• 一个模拟 CPU 重任务
• 过载时返回 503

项目结构如下：

.
├── server.js
├── worker-pool.js
└── task-worker.js

1. Worker 执行文件

这里用“计算斐波那契数”模拟 CPU 密集型任务。这个例子虽然有点老派，但好处是简单直观，方便观察线程效果。

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

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

parentPort.on('message', (message) => {
  const { taskId, payload } = message;
  const start = Date.now();

  try {
    const { n } = payload;

    if (!Number.isInteger(n) || n < 0 || n > 45) {
      throw new Error('参数 n 必须是 0~45 的整数');
    }

    const result = fib(n);
    const duration = Date.now() - start;

    parentPort.postMessage({
      taskId,
      ok: true,
      result,
      duration
    });
  } catch (error) {
    parentPort.postMessage({
      taskId,
      ok: false,
      error: error.message
    });
  }
});

2. 实现 Worker Pool

这个版本实现了：

• 固定数量 worker
• 空闲 worker 管理
• 任务队列
• 单任务超时
• worker 异常退出自动重建

// worker-pool.js
const path = require('path');
const { Worker } = require('worker_threads');
const os = require('os');

class WorkerPool {
  constructor(options = {}) {
    this.size = options.size || Math.max(1, os.cpus().length - 1);
    this.maxQueue = options.maxQueue || this.size * 20;
    this.taskTimeout = options.taskTimeout || 10000;
    this.workerFile = options.workerFile || path.resolve(__dirname, 'task-worker.js');

    this.workers = [];
    this.idleWorkers = [];
    this.queue = [];
    this.taskId = 0;
    this.pendingTasks = new Map();

    for (let i = 0; i < this.size; i++) {
      this._createWorker();
    }
  }

  _createWorker() {
    const worker = new Worker(this.workerFile);
    worker.currentTaskId = null;

    worker.on('message', (message) => {
      const { taskId, ok, result, error, duration } = message;
      const task = this.pendingTasks.get(taskId);
      if (!task) return;

      clearTimeout(task.timeoutId);
      this.pendingTasks.delete(taskId);
      worker.currentTaskId = null;
      this.idleWorkers.push(worker);

      if (ok) {
        task.resolve({ result, duration });
      } else {
        task.reject(new Error(error));
      }

      this._dispatch();
    });

    worker.on('error', (err) => {
      if (worker.currentTaskId) {
        const task = this.pendingTasks.get(worker.currentTaskId);
        if (task) {
          clearTimeout(task.timeoutId);
          this.pendingTasks.delete(worker.currentTaskId);
          task.reject(err);
        }
      }
    });

    worker.on('exit', (code) => {
      this.workers = this.workers.filter((w) => w !== worker);
      this.idleWorkers = this.idleWorkers.filter((w) => w !== worker);

      if (worker.currentTaskId) {
        const task = this.pendingTasks.get(worker.currentTaskId);
        if (task) {
          clearTimeout(task.timeoutId);
          this.pendingTasks.delete(worker.currentTaskId);
          task.reject(new Error(`Worker exited with code ${code}`));
        }
      }

      this._createWorker();
      this._dispatch();
    });

    this.workers.push(worker);
    this.idleWorkers.push(worker);
  }

  execute(payload) {
    if (this.queue.length >= this.maxQueue) {
      return Promise.reject(new Error('任务队列已满'));
    }

    const taskId = ++this.taskId;

    return new Promise((resolve, reject) => {
      this.queue.push({ taskId, payload, resolve, reject, enqueueTime: Date.now() });
      this._dispatch();
    });
  }

  _dispatch() {
    while (this.idleWorkers.length > 0 && this.queue.length > 0) {
      const worker = this.idleWorkers.shift();
      const task = this.queue.shift();

      worker.currentTaskId = task.taskId;

      const timeoutId = setTimeout(() => {
        this.pendingTasks.delete(task.taskId);
        task.reject(new Error('任务执行超时'));

        try {
          worker.terminate();
        } catch (e) {}
      }, this.taskTimeout);

      this.pendingTasks.set(task.taskId, {
        ...task,
        timeoutId
      });

      worker.postMessage({
        taskId: task.taskId,
        payload: task.payload
      });
    }
  }

  getStats() {
    return {
      size: this.size,
      busyWorkers: this.workers.length - this.idleWorkers.length,
      idleWorkers: this.idleWorkers.length,
      queueLength: this.queue.length,
      pendingTasks: this.pendingTasks.size
    };
  }

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

module.exports = WorkerPool;

3. HTTP 服务与事件循环监控

这里我们用 Node 原生 http 模块，避免引入框架噪音，重点看架构本身。

// server.js
const http = require('http');
const { URL } = require('url');
const { monitorEventLoopDelay } = require('perf_hooks');
const os = require('os');
const WorkerPool = require('./worker-pool');

const pool = new WorkerPool({
  size: Math.max(1, os.cpus().length - 1),
  maxQueue: 100,
  taskTimeout: 15000
});

const histogram = monitorEventLoopDelay({ resolution: 20 });
histogram.enable();

function sendJson(res, statusCode, data) {
  res.writeHead(statusCode, {
    'Content-Type': 'application/json; charset=utf-8'
  });
  res.end(JSON.stringify(data));
}

function isOverloaded() {
  const stats = pool.getStats();
  const meanMs = Number(histogram.mean / 1e6).toFixed(2);
  const p99Ms = Number(histogram.percentile(99) / 1e6).toFixed(2);

  return {
    overloaded: stats.queueLength > 80 || Number(p99Ms) > 100,
    stats,
    eventLoop: {
      meanMs: Number(meanMs),
      p99Ms: Number(p99Ms)
    }
  };
}

const server = http.createServer(async (req, res) => {
  const url = new URL(req.url, `http://${req.headers.host}`);

  if (req.method === 'GET' && url.pathname === '/health') {
    const status = isOverloaded();
    return sendJson(res, 200, {
      ok: true,
      ...status
    });
  }

  if (req.method === 'GET' && url.pathname === '/compute') {
    const status = isOverloaded();

    if (status.overloaded) {
      return sendJson(res, 503, {
        ok: false,
        message: '服务繁忙，请稍后重试',
        ...status
      });
    }

    const n = Number(url.searchParams.get('n') || 40);

    try {
      const begin = Date.now();
      const result = await pool.execute({ n });
      const totalDuration = Date.now() - begin;

      return sendJson(res, 200, {
        ok: true,
        input: n,
        workerDuration: result.duration,
        totalDuration,
        result: result.result,
        pool: pool.getStats()
      });
    } catch (error) {
      return sendJson(res, 500, {
        ok: false,
        message: error.message
      });
    }
  }

  sendJson(res, 404, {
    ok: false,
    message: 'Not Found'
  });
});

server.listen(3000, () => {
  console.log('server started at http://localhost:3000');
  console.log('try: http://localhost:3000/compute?n=40');
  console.log('health: http://localhost:3000/health');
});

process.on('SIGINT', async () => {
  console.log('\nshutting down...');
  await pool.destroy();
  process.exit(0);
});

4. 运行方式

node server.js

访问：

curl "http://localhost:3000/compute?n=40"
curl "http://localhost:3000/health"

也可以简单压测一下：

seq 1 20 | xargs -I {} -P 10 curl -s "http://localhost:3000/compute?n=40" > /dev/null

5. 请求处理时序图

这张图能帮助你把“请求、排队、调度、执行、回传”串起来。

sequenceDiagram
    participant C as Client
    participant M as Main Thread
    participant Q as Task Queue
    participant W as Worker

    C->>M: GET /compute?n=40
    M->>M: 检查事件循环延迟/队列长度
    alt 系统过载
        M-->>C: 503 服务繁忙
    else 可接收
        M->>Q: 入队任务
        M->>W: 分配空闲 Worker
        W->>W: 执行 CPU 计算
        W-->>M: 返回结果
        M-->>C: 200 + result
    end

核心代码背后的设计点

光把代码跑起来还不够，真正实战时，下面这些点很关键。

1. 为什么健康检查不能只返回“进程还活着”

如果健康检查只是：

res.end('ok')

那它只能说明进程没死，不能说明服务还能扛请求。

更有意义的健康状态应该包含：

• 队列长度
• busy worker 数
• 事件循环 p95 / p99 延迟
• 是否触发过载阈值

这样你在网关、K8s、SLB 后面做流量摘除时，判断才靠谱。

2. 为什么拒绝新任务比无限排队更健康

很多业务团队会天然抗拒 503，觉得“失败率不好看”。但从系统角度看：

• 快速失败 是可控的
• 长时间排队后超时失败 是不可控的

后者不仅用户体验更差，还会占着连接、占着内存、拖累整机。

所以高并发服务里，一个常识是：宁可明确拒绝，也不要悄悄堆积。

3. 为什么任务参数要做上限限制

示例里把 n 限制在 45 以内，不是随便写的。因为有些 CPU 任务复杂度是指数级的，一旦你不做限制，请求参数本身就会成为 DoS 入口。

这个问题在以下任务里尤其常见：

• 正则表达式灾难性回溯
• 大对象深度递归
• 大规模排序或组合运算
• 无上限压缩/解压

常见坑与排查

这部分我尽量写得贴近真实问题，而不是停留在“注意性能”这种空话上。

坑 1：Worker 开太多，吞吐反而下降

现象

• CPU 100%
• 上下文切换明显变多
• 响应时间不降反升
• 系统 load 很高

原因

线程数超过 CPU 实际承载能力后，会进入频繁抢占。尤其任务都很“纯计算”时，线程越多不一定越快。

排查方法

• 对比不同 pool size 的吞吐和 p99
• 观察系统层面的 CPU 与 load
• 看事件循环延迟是否也被带高

建议

先从 cpu核数 - 1 开始压测，而不是拍脑袋设成 32 或 64。

坑 2：消息传递对象过大，主线程照样卡

现象

已经用了 Worker，但主线程仍然卡顿，事件循环延迟高。

原因

Worker 与主线程传消息时，复杂对象可能带来明显的序列化/拷贝成本。比如你把几十 MB 的对象来回传，主线程还是会吃不消。

排查方法

• 打印请求体大小、消息对象大小
• 用 process.memoryUsage() 看堆内存是否异常增长
• 观察“worker 计算时间很短，但总耗时很长”的情况

建议

• 尽量只传必要字段
• 大 Buffer 考虑 Transferable
• 更极端场景考虑 SharedArrayBuffer
• 结果不要返回冗余中间态数据

坑 3：只监控 CPU，不监控事件循环

现象

CPU 看起来没满，但服务还是慢、超时多。

原因

主线程可能被 GC、序列化、大量同步逻辑等卡住，而这些问题不总是直接等同于 CPU 满载。

排查方法

重点看：

• monitorEventLoopDelay() 的 p95 / p99
• 请求总耗时与 worker 内部耗时的差值
• 主线程是否存在同步日志、同步 JSON 大对象处理

建议

把事件循环延迟当成一等公民指标。

坑 4：任务超时后没清理干净，导致资源泄漏

现象

跑一段时间后：

• 内存上涨
• pending task 数量不下降
• worker 数异常
• 队列越来越慢

原因

典型问题有：

• 超时后没删除 pendingTasks
• worker 卡死后没重建
• Promise 只 reject 不清资源
• terminate 后仍保留引用

排查方法

• 周期性打印池状态
• 压测后观察 pendingTasks 是否归零
• 检查 exit/error/message 三条链路是否都收口

建议

任务生命周期一定要“有始有终”，无论成功、失败、超时、崩溃，都要回收。

坑 5：把 Worker Threads 当成分布式任务系统

现象

需求慢慢变成：

• 任务要持久化
• 服务重启后任务不能丢
• 失败要重试
• 支持延迟任务、优先级任务
• 多机共享消费

原因

这已经超出单进程 worker pool 的边界了。

建议

当你出现这些需求时，应该考虑：

• Redis Stream / BullMQ
• RabbitMQ / Kafka
• 独立消费者服务

Worker Threads 很强，但它不是消息队列。

安全/性能最佳实践

这一节我把“能直接带走用”的建议列得更明确一些。

1. 参数校验必须前置

不要把非法请求直接扔进 Worker 再说。主线程先做轻量校验：

• 类型检查
• 长度限制
• 范围限制
• 白名单校验

这样可以减少无意义排队与线程占用。

function validateN(n) {
  return Number.isInteger(n) && n >= 0 && n <= 45;
}

2. 设定明确的过载阈值

至少要有两个阈值：

• 队列长度阈值
• 事件循环延迟阈值

例如：

queueLength > 80 触发拒绝
eventLoop p99 > 100ms 触发降级

阈值不是标准答案，必须结合压测结果调整。

3. 区分“同步返回结果”和“异步接单”

如果任务执行时间已经明显超过接口可接受时延，比如 5 秒、10 秒以上，那就别坚持同步 HTTP 返回了。

更好的模式是：

1. 提交任务
2. 返回任务 ID
3. 异步查询结果或回调通知

Worker Pool 能解决并行计算，但不能改变用户等待耐心。

4. 给 Worker 设置崩溃恢复机制

Worker 可能因代码异常、内存问题退出，所以线程池要能：

• 自动重建 worker
• 清理关联任务
• 记录异常日志
• 上报告警

示例里已经实现了退出后自动重建，这是生产可用性的基础动作。

5. 日志不要把大对象全量打出来

我见过不少“排查问题时把 payload 全打印”的做法，结果日志本身成了新的性能瓶颈。

建议：

• 打 taskId、耗时、结果摘要
• 对大对象做截断
• 错误日志记录关键信息，不要无脑全量 dump

6. 结合指标做闭环

至少采集这些指标：

• 请求 QPS
• 成功率 / 503 比例
• 平均耗时 / p95 / p99
• worker busy 数
• 队列长度
• 事件循环延迟
• 进程 RSS / heapUsed

如果没有这些指标，所谓“调优”往往只是感觉。

7. 尽量避免在主线程做重序列化

例如：

• 大型 JSON.stringify
• 深拷贝
• 大数组 map/filter/reduce 链式处理
• 同步压缩/加密

这些都可能让你“明明用了 Worker，主线程还在抖”。

一个更完整的状态视角

为了更容易排查系统在不同阶段的表现，可以把任务状态理解成下面这样：

stateDiagram-v2
    [*] --> Queued
    Queued --> Running: 分配到空闲 Worker
    Queued --> Rejected: 队列已满/系统过载
    Running --> Success: 正常完成
    Running --> Failed: 业务异常
    Running --> Timeout: 执行超时
    Running --> WorkerExit: Worker 崩溃退出
    Timeout --> [*]
    Failed --> [*]
    Success --> [*]
    Rejected --> [*]
    WorkerExit --> [*]

把这些状态在日志和监控上区分开，排障效率会高很多。否则最后你只能看到一句“任务失败”，却不知道到底是：

• 被拒绝了
• 排队太久了
• worker 崩了
• 还是参数本身非法

什么时候该升级架构

这套方案很适合：

• 单机多核场景
• CPU 密集型任务
• 希望在现有 Node 服务内部完成并行处理
• 对“立即返回结果”还有需求

但如果你遇到下面情况，就要考虑升级：

1. 任务必须持久化

只靠内存队列，进程一挂任务就没了。此时应引入外部任务队列。

2. 任务执行时间很长

如果动不动几十秒、几分钟，HTTP 同步返回就不合理了，应改为异步任务模式。

3. 需要跨机器扩缩容

单机 worker pool 无法天然跨实例共享任务，需要 MQ 或统一调度系统。

4. 任务优先级和重试策略复杂

比如：

• VIP 任务优先
• 失败按指数退避重试
• 死信队列
• 人工补偿

这些都更像完整任务平台，而不是单服务线程池。

总结

如果把这篇文章浓缩成一句话，那就是：

在 Node.js 里做高并发 CPU 任务处理，关键不是“把计算放进 Worker”这么简单，而是“用 Worker Pool + 事件循环监控 + 过载保护”一起设计。

你真正要带走的实践建议有这几条：

1. 主线程只做接入、校验、调度，不做重计算
2. 使用固定大小 Worker Pool，不要按请求临时创建 Worker
3. 同时监控队列长度和事件循环延迟
4. 过载时快速失败，不要无限堆积
5. 给任务加超时、给 Worker 加重建、给系统加指标
6. 明确这套方案的边界：它是单机并行方案，不是分布式任务平台

如果你现在的 Node 服务已经出现“CPU 一高，整个接口都慢”的问题，那么最值得先做的不是重写全部系统，而是先落地下面这个最小闭环：

• 一个线程池
• 一个排队上限
• 一个 /health
• 一个事件循环延迟指标
• 一个清晰的拒绝策略

这套东西不花哨，但非常管用。很多服务稳定性提升，恰恰不是来自更复杂的框架，而是来自这种朴素但成体系的控制点。