背景与问题

很多同学第一次做 Node.js 服务时，会天然觉得“Node 很快，所以高并发没问题”。这句话只对了一半。

Node.js 在 I/O 密集型场景 下确实非常能打，比如网关转发、接口聚合、数据库读写编排。但一旦业务里混入 CPU 密集型任务，事情就变了：

图片处理
大量 JSON 计算/压缩
报表聚合
加密/解密
文件解析
批量规则匹配

这类任务如果直接跑在主线程里，会阻塞事件循环。结果通常很熟悉：

接口延迟突然升高
偶发超时
CPU 打满但吞吐并不高
健康检查失败，容器被重启
明明机器配置不低，服务还是“不抗压”

我以前就踩过一个典型坑：一个“导出报表”接口，把几十万行数据聚合后再做格式转换，逻辑本身没问题，但它直接跑在 Node 主线程。线上高峰时，一边是导出任务在狂吃 CPU，一边普通查询接口开始排队，最后整个实例看起来像“活着”，实际上已经接近不可用。

这时，Worker Threads + 任务队列 就是很务实的一套解法：

主线程只负责接请求、入队、返回任务状态
Worker Threads 专门消费 CPU 任务
队列负责削峰、限流、调度、重试
服务从“谁来都一起抢 CPU”变成“有秩序地处理任务”

这篇文章就从架构角度，带你搭一个 可运行、可扩展、便于排查 的高并发任务处理服务。

方案概览：为什么是 Worker Threads + 队列

先给结论：

只用异步 Promise：解决不了 CPU 阻塞
只开多个 Node 进程：可以横向分担，但单进程内部仍可能被重任务拖垮
只用 Worker Threads：能并行计算，但没有调度层，容易把线程和内存打爆
Worker Threads + 队列：既能并行，又能控速，还能管理任务生命周期

整体架构

flowchart LR
    A[客户端请求] --> B[HTTP API / 主线程]
    B --> C[任务队列 Queue]
    C --> D[调度器 Scheduler]
    D --> E1[Worker 1]
    D --> E2[Worker 2]
    D --> E3[Worker N]
    E1 --> F[结果存储 Result Map / DB]
    E2 --> F
    E3 --> F
    B --> F
    F --> G[状态查询接口]

这里的关键不是“把任务扔进线程里”，而是加上两层控制：

队列层：控制任务顺序、容量、重试
调度层：控制并发 Worker 数量，避免瞬时挤爆 CPU

核心原理

1. 主线程为什么会被 CPU 任务拖慢

Node.js 的 JavaScript 执行主要跑在主线程事件循环中。I/O 可以异步，但如果你的 JS 在做大计算，比如：

for (let i = 0; i < 1e9; i++) {
  // 纯 CPU 运算
}

事件循环在这段逻辑结束前无法继续处理新的请求、定时器、回调。于是整个服务表现为：

“不是挂了，但也不响应”
TCP 连接在，但业务超时
延迟抖动非常严重

2. Worker Threads 的角色

worker_threads 是 Node.js 提供的真正多线程能力。它适合把 CPU 密集型 JS 任务 放到独立线程中执行。

特点：

与主线程并行执行
可通过消息通信传递任务和结果
适合计算、转换、解析等逻辑
不适合无限制创建，线程过多会适得其反

3. 队列机制的价值

如果来了 5000 个重任务，直接创建 5000 个 Worker 显然不现实。队列的意义是：

削峰：请求先排队
限流：只让固定数量任务同时运行
背压：队列过长时拒绝新任务或降级
可观测：知道有多少待处理、运行中、成功、失败任务

4. 线程池比“来一个任务开一个线程”更靠谱

线程创建有成本，频繁创建/销毁会浪费资源。更合理的方式是：

启动时初始化固定大小的 Worker 池
任务进入队列
空闲 Worker 从队列中取任务执行
执行完成后回到空闲状态

sequenceDiagram
    participant Client
    participant API as Main Thread API
    participant Queue
    participant Scheduler
    participant Worker

    Client->>API: POST /tasks
    API->>Queue: 入队任务
    API-->>Client: 返回 taskId
    Scheduler->>Queue: 拉取待执行任务
    Queue-->>Scheduler: task
    Scheduler->>Worker: 分发任务
    Worker-->>Scheduler: 执行结果
    Scheduler->>API: 更新任务状态
    Client->>API: GET /tasks/:id
    API-->>Client: 返回状态/结果

方案对比与取舍分析

方案一：直接在主线程执行

优点

最简单
没有跨线程通信成本

缺点

CPU 任务会阻塞整个服务
高并发下延迟不可控
基本不适合线上重计算接口

适用场景

小工具
低频任务
脚本型程序

方案二：Cluster / 多进程

优点

能利用多核
部署简单

缺点

单个进程内部仍会被重任务阻塞
任务调度和状态管理要自己做
跨进程共享状态更麻烦

适用场景

主要是 Web 服务扩容
并不专门针对重计算任务

方案三：消息队列 + 独立消费服务

优点

解耦彻底
可水平扩展
适合大型系统

缺点

引入 Redis/RabbitMQ/Kafka 等基础设施
开发运维复杂度上升

适用场景

多服务协作
任务持久化要求高
需要跨实例共享任务

方案四：Worker Threads + 进程内队列

优点

本地即可运行
比较适合中型业务
能快速提升 CPU 任务吞吐

缺点

队列不持久化，进程重启任务会丢
只适合单实例或轻量场景
超大规模下仍需外部消息系统

适用场景

单服务内的计算任务
中等规模高并发处理
需要先验证架构收益

容量估算：线程数和队列长度怎么定

这部分很容易被忽略，但实际上非常重要。

Worker 数量建议

一般不要直接等于 CPU 核数，更不要盲目翻倍。经验上可以从：

workerCount = max(1, CPU核心数 - 1)

开始压测，再看：

主线程是否还有余量处理网络请求
每个 Worker 的内存占用
单任务平均 CPU 时间
上下文切换是否明显升高

如果你的任务特别“吃 CPU”，我通常建议先从 CPU 核数 - 1 起步。比如 8 核机器先开 6~7 个 Worker，不要一上来 16 个。

队列长度建议

队列不是越长越好。太长意味着：

响应变慢
内存占用上升
用户等待不可接受
故障恢复时间变长

可以先设一个上限，比如 1000 或 5000，超过后直接返回：

429 Too Many Requests
或业务态“系统繁忙，请稍后重试”

实战代码（可运行）

下面我们实现一个最小可用版本：

Express 提供 HTTP 接口
内存队列保存任务
Worker 线程池执行 CPU 密集型任务
提供任务提交和状态查询接口

项目结构如下：

task-service/
  ├─ package.json
  ├─ server.js
  └─ worker.js

1. 安装依赖

npm init -y
npm install express

2. worker.js

这里用一个故意耗 CPU 的计算任务模拟实际场景，比如复杂数据计算。

const { parentPort } = require('worker_threads');

function heavyCompute(n) {
  let count = 0;
  for (let i = 2; i <= n; i++) {
    let isPrime = true;
    const limit = Math.sqrt(i);
    for (let j = 2; j <= limit; j++) {
      if (i % j === 0) {
        isPrime = false;
        break;
      }
    }
    if (isPrime) count++;
  }
  return count;
}

parentPort.on('message', (task) => {
  try {
    const { taskId, payload } = task;
    const start = Date.now();
    const result = heavyCompute(payload.n);
    const duration = Date.now() - start;

    parentPort.postMessage({
      taskId,
      status: 'completed',
      result: {
        primeCount: result,
        duration,
      },
    });
  } catch (error) {
    parentPort.postMessage({
      taskId: task.taskId,
      status: 'failed',
      error: error.message,
    });
  }
});

3. server.js

const express = require('express');
const os = require('os');
const path = require('path');
const { Worker } = require('worker_threads');
const crypto = require('crypto');

const app = express();
app.use(express.json());

const CPU_COUNT = os.cpus().length;
const WORKER_COUNT = Math.max(1, CPU_COUNT - 1);
const MAX_QUEUE_SIZE = 1000;

const taskQueue = [];
const taskStore = new Map();
const workers = [];
const idleWorkers = new Set();

function createTaskId() {
  return crypto.randomUUID();
}

function updateTask(taskId, patch) {
  const oldTask = taskStore.get(taskId);
  if (!oldTask) return;
  taskStore.set(taskId, { ...oldTask, ...patch, updatedAt: Date.now() });
}

function dispatchTasks() {
  while (taskQueue.length > 0 && idleWorkers.size > 0) {
    const worker = idleWorkers.values().next().value;
    idleWorkers.delete(worker);

    const task = taskQueue.shift();
    updateTask(task.taskId, {
      status: 'running',
      startedAt: Date.now(),
      workerId: worker.workerId,
    });

    worker.currentTaskId = task.taskId;
    worker.postMessage(task);
  }
}

function createWorker(index) {
  const worker = new Worker(path.resolve(__dirname, './worker.js'));
  worker.workerId = `worker-${index}`;
  worker.currentTaskId = null;

  worker.on('message', (message) => {
    const { taskId, status, result, error } = message;

    if (status === 'completed') {
      updateTask(taskId, {
        status: 'completed',
        result,
        error: null,
        finishedAt: Date.now(),
      });
    } else {
      updateTask(taskId, {
        status: 'failed',
        error,
        finishedAt: Date.now(),
      });
    }

    worker.currentTaskId = null;
    idleWorkers.add(worker);
    dispatchTasks();
  });

  worker.on('error', (err) => {
    console.error(`[${worker.workerId}] error:`, err);

    if (worker.currentTaskId) {
      updateTask(worker.currentTaskId, {
        status: 'failed',
        error: `Worker error: ${err.message}`,
        finishedAt: Date.now(),
      });
    }

    idleWorkers.delete(worker);
    workers.splice(workers.indexOf(worker), 1);

    // 拉起新 worker，保持线程池规模
    const newWorker = createWorker(index);
    workers.push(newWorker);
    idleWorkers.add(newWorker);
    dispatchTasks();
  });

  worker.on('exit', (code) => {
    console.log(`[${worker.workerId}] exited with code ${code}`);
  });

  return worker;
}

// 初始化 worker 池
for (let i = 0; i < WORKER_COUNT; i++) {
  const worker = createWorker(i);
  workers.push(worker);
  idleWorkers.add(worker);
}

// 健康检查
app.get('/health', (req, res) => {
  res.json({
    ok: true,
    cpuCount: CPU_COUNT,
    workerCount: WORKER_COUNT,
    queueSize: taskQueue.length,
    idleWorkerCount: idleWorkers.size,
    totalTasks: taskStore.size,
  });
});

// 提交任务
app.post('/tasks', (req, res) => {
  const n = Number(req.body?.n);

  if (!Number.isInteger(n) || n < 2 || n > 200000) {
    return res.status(400).json({
      error: '参数 n 必须是 2~200000 之间的整数',
    });
  }

  if (taskQueue.length >= MAX_QUEUE_SIZE) {
    return res.status(429).json({
      error: '任务队列已满，请稍后重试',
    });
  }

  const taskId = createTaskId();
  const now = Date.now();

  const task = {
    taskId,
    payload: { n },
  };

  taskStore.set(taskId, {
    taskId,
    status: 'queued',
    payload: { n },
    result: null,
    error: null,
    createdAt: now,
    updatedAt: now,
    startedAt: null,
    finishedAt: null,
    workerId: null,
  });

  taskQueue.push(task);
  dispatchTasks();

  res.status(202).json({
    taskId,
    status: 'queued',
  });
});

// 查询单个任务
app.get('/tasks/:taskId', (req, res) => {
  const task = taskStore.get(req.params.taskId);

  if (!task) {
    return res.status(404).json({
      error: '任务不存在',
    });
  }

  res.json(task);
});

// 查看队列概要
app.get('/metrics', (req, res) => {
  let queued = 0;
  let running = 0;
  let completed = 0;
  let failed = 0;

  for (const task of taskStore.values()) {
    if (task.status === 'queued') queued++;
    else if (task.status === 'running') running++;
    else if (task.status === 'completed') completed++;
    else if (task.status === 'failed') failed++;
  }

  res.json({
    queueSize: taskQueue.length,
    idleWorkerCount: idleWorkers.size,
    workerCount: workers.length,
    taskStats: {
      queued,
      running,
      completed,
      failed,
    },
  });
});

const PORT = 3000;
app.listen(PORT, () => {
  console.log(`Task service listening on http://localhost:${PORT}`);
});

4. 启动服务

node server.js

5. 提交任务测试

curl -X POST http://localhost:3000/tasks \
  -H "Content-Type: application/json" \
  -d '{"n": 80000}'

返回示例：

{
  "taskId": "b0e4a17e-9e0a-44f4-a879-1d8f79f6b5f8",
  "status": "queued"
}

然后查询状态：

curl http://localhost:3000/tasks/b0e4a17e-9e0a-44f4-a879-1d8f79f6b5f8

6. 任务生命周期

stateDiagram-v2
    [*] --> queued
    queued --> running
    running --> completed
    running --> failed
    failed --> [*]
    completed --> [*]

这套实现解决了什么问题

到这里，主线程做的事只有：

接收请求
参数校验
任务入队
状态查询
结果返回

真正耗 CPU 的逻辑被扔给 Worker。这样即使同时有多个重任务，主线程仍然可以较稳定地响应新请求。

这和“异步函数”是两回事。async/await 只是不阻塞 I/O 等待，但 不会让 CPU 计算自动并行。而 Worker Threads 才是把计算转移到其他线程。

常见坑与排查

这一节我尽量讲得接地气一点，因为真正上线后，问题通常不是“代码不会写”，而是“为什么它跑着跑着不对劲”。

1. 误把异步当并行

很多代码看起来是这样的：

await Promise.all(tasks.map(task => heavyCompute(task)));

这不会让 CPU 任务并行到多个线程。它只是把多个计算安排在当前线程里执行，最终仍然会阻塞主线程。

排查方式：

观察接口延迟是否随计算任务升高
看单进程 CPU 是否接近 100%
用 clinic doctor 或 0x 做事件循环分析

2. 队列无限增长

如果消费者速度跟不上生产速度，内存队列会越来越长，最终把进程拖死。

现象：

RSS 持续上涨
GC 频繁
延迟升高
最后 OOM

处理建议：

给队列设置上限
对调用方返回 429
根据优先级丢弃低价值任务
超过 SLA 的任务直接取消

3. Worker 异常退出后任务丢失

Worker 可能因为代码异常、资源不足或未捕获错误退出。如果不处理，当前任务状态就会一直卡在 running。

正确做法：

记录 worker.currentTaskId
error 或 exit 时把该任务标记为失败或重新入队
补拉起新的 Worker

我们上面的代码已经做了最基本的失败标记与自动补 Worker，但真实业务里更建议加入 重试次数。

4. 结果对象过大，跨线程通信变慢

Worker 和主线程通过消息传递通信。你如果把超大对象来回复制，通信成本会很高。

建议：

传递最小必要参数
大文件传路径或对象存储地址，不要直接传文件内容
对二进制可考虑 Transferable / ArrayBuffer

5. 主线程 Map 一直存任务，内存泄漏

示例中为了简单，用 Map 持久保存任务状态。但线上如果一直不清理，迟早会出问题。

建议：

给任务结果设置 TTL
完成后写入数据库或 Redis
定时清理历史任务

例如可以加一个简化清理器：

setInterval(() => {
  const now = Date.now();
  const ttl = 10 * 60 * 1000;

  for (const [taskId, task] of taskStore.entries()) {
    if (
      ['completed', 'failed'].includes(task.status) &&
      task.finishedAt &&
      now - task.finishedAt > ttl
    ) {
      taskStore.delete(taskId);
    }
  }
}, 60 * 1000);

6. 队列顺序不等于业务优先级

FIFO 很简单，但不一定合理。比如：

用户前台导出任务
后台批处理任务

如果都用一个队列，低优先级大任务可能会堵住高优先级小任务。

建议：

引入多级队列
按优先级调度
给紧急任务预留并发槽位

安全/性能最佳实践

1. 参数校验必须前置

CPU 任务很怕“恶意输入”。别人传一个特别大的参数，就可能消耗大量 CPU 时间。

建议至少做：

类型校验
数值范围限制
请求体大小限制
鉴权和限流

Express 可以配合中间件做更细粒度限制。

2. 队列要有背压策略

背压不是“高级功能”，而是高并发系统的基本修养。至少要明确三件事：

队列最大长度是多少
满了以后是拒绝、降级还是转储
调用方收到失败后如何重试

如果业务允许，我更推荐 明确拒绝，而不是“先收下再慢慢处理”，因为后者会把问题延后并放大。

3. Worker 数量不要拍脑袋

不是线程越多越快。线程过多会带来：

上下文切换增加
内存变大
主线程资源被挤压
总吞吐反而下降

一定要通过压测确定最佳并发数。

4. 给任务设置超时

有些计算逻辑在特定输入下会异常慢。建议给任务加超时控制。

一种简单做法是在主线程记录开始时间，如果超过阈值就标记失败并销毁 Worker，再补一个新 Worker。

伪代码如下：

function withTaskTimeout(worker, taskId, timeoutMs) {
  const timer = setTimeout(() => {
    console.error(`task ${taskId} timeout`);
    worker.terminate();
  }, timeoutMs);

  return () => clearTimeout(timer);
}

5. 可观测性要先于“优化”

没有指标就别谈优化。至少监控这些数据：

队列长度
排队时间
任务执行时间
Worker 忙闲比例
失败率
进程 CPU / 内存
事件循环延迟

如果你在线上只看到“CPU 高”，但不知道是队列积压、任务变慢还是某个 Worker 异常，那排查会非常被动。

6. 结果持久化与幂等设计

本文示例是进程内内存存储，适合演示，不适合强可靠场景。真正上线时建议：

队列进 Redis / RabbitMQ / Kafka
任务状态存 Redis / MySQL
任务处理做幂等
支持重复投递去重

边界条件要想清楚：

服务重启后未完成任务怎么办？
调用方重复提交怎么办？
任务已经完成但客户端超时重试怎么办？

这些都属于架构能否稳定落地的关键点。

进一步演进建议

如果你的业务量继续增长，可以按这个路径升级：

阶段一：单机内存队列 + Worker 池

适合：

快速验证收益
中小规模单实例服务
可接受任务短暂丢失

阶段二：Redis 队列 + 多实例 Worker 服务

适合：

多实例部署
需要任务持久化
支持弹性扩缩容

阶段三：任务平台化

特征包括：

多优先级队列
重试和死信队列
任务超时与取消
指标和告警
租户隔离
配额控制

你会发现，到了这个阶段，系统已经从“写个接口”变成“任务处理平台”。

总结

对于 Node.js 来说，真正难的不是“并发请求多”，而是 高并发下混入 CPU 密集型任务。这时候如果继续把计算堆在主线程里，服务很容易出现高延迟、抖动甚至雪崩。

一个比较平衡、也很适合中级开发者上手的方案是：

用 Worker Threads 承担 CPU 计算
用 任务队列 做削峰、限流和调度
用 固定大小线程池 控制资源消耗
用 状态存储和监控指标 保证可排查

如果你准备把它用到真实项目里，我的建议是按这个顺序落地：

先把 CPU 任务从主线程剥离出去
再加队列上限和 429 背压
然后补任务超时、失败重试、历史清理
最后再考虑 Redis 持久化和多实例扩容

一句话收尾：Worker Threads 解决的是“能并行算”，队列机制解决的是“怎么稳稳地算”。两者结合，Node.js 才更像一个能扛住高并发重任务的服务端运行时。

《Node.js 中级实战：基于 Worker Threads 与队列机制构建高并发任务处理服务》

背景与问题

方案概览：为什么是 Worker Threads + 队列

整体架构

核心原理

1. 主线程为什么会被 CPU 任务拖慢

2. Worker Threads 的角色

3. 队列机制的价值

4. 线程池比“来一个任务开一个线程”更靠谱

方案对比与取舍分析

方案一：直接在主线程执行

方案二：Cluster / 多进程

方案三：消息队列 + 独立消费服务

方案四：Worker Threads + 进程内队列

容量估算：线程数和队列长度怎么定

Worker 数量建议

队列长度建议

实战代码（可运行）

1. 安装依赖

2. worker.js

3. server.js

4. 启动服务

5. 提交任务测试

6. 任务生命周期

这套实现解决了什么问题

常见坑与排查

1. 误把异步当并行

2. 队列无限增长

3. Worker 异常退出后任务丢失

4. 结果对象过大，跨线程通信变慢

5. 主线程 Map 一直存任务，内存泄漏

6. 队列顺序不等于业务优先级

安全/性能最佳实践

1. 参数校验必须前置

2. 队列要有背压策略

3. Worker 数量不要拍脑袋

4. 给任务设置超时

5. 可观测性要先于“优化”

6. 结果持久化与幂等设计

进一步演进建议

阶段一：单机内存队列 + Worker 池

阶段二：Redis 队列 + 多实例 Worker 服务

阶段三：任务平台化

总结