背景与问题

Node.js 很擅长处理高并发 I/O，但一旦任务里混入 CPU 密集型计算，事情就开始变味了。

比如这些场景：

图片压缩、音视频转码
大批量数据清洗
报表聚合计算
风控规则批处理
文本分词、摘要、向量化预处理

如果你把这类任务直接塞进主线程，即使接口本身写得再优雅，也会遇到几个很现实的问题：

事件循环被阻塞
某个计算任务跑满 CPU，整个服务的响应时间会突然抖高。
请求型系统和任务型系统混在一起
用户请求希望“快返回”，后台任务却可能“慢慢算”。两种诉求天然冲突。
任务洪峰不可控
比如一分钟内涌入 5 万个任务，如果没有缓冲层，应用实例只会被瞬间打爆。
失败重试和幂等处理复杂
任务执行失败后，怎么重试？重复消费怎么办？执行一半进程挂了怎么办？

所以，比较稳妥的架构通常不是“主线程硬扛”，而是：

主服务负责接收任务
消息队列负责削峰填谷
Worker Threads 负责并行执行 CPU 密集型任务

这篇文章我会从架构角度，带你搭一个可运行的 Node.js 示例，重点不是“会不会用 API”，而是“为什么这么拆，以及拆完后怎么稳”。

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

先给结论：

Worker Threads 解决的是：单进程内如何把 CPU 密集型工作从主线程挪走
消息队列 解决的是：任务如何解耦、削峰、重试、异步化
两者结合，才更像一个能在生产环境站得住的高并发任务处理方案

一个典型链路

flowchart LR
    A[客户端/业务系统] --> B[Node.js API 服务]
    B --> C[消息队列]
    C --> D[任务消费者 Consumer]
    D --> E[Worker Pool]
    E --> F[任务结果存储/数据库]
    E --> G[日志/监控]

这里的职责边界很关键：

API 服务：收任务、校验参数、快速返回任务 ID
消息队列：缓存任务、平滑流量、支持确认机制
Consumer：从队列拉取任务，控制消费速率
Worker Pool：真正做 CPU 密集计算
结果存储：记录任务状态、结果、失败原因

如果没有消息队列，任务一多，入口服务和执行服务会直接耦合；
如果没有 Worker Threads，消费者拿到任务后仍然可能阻塞自己的事件循环。

核心原理

1. Worker Threads 不是“多进程”，而是“同进程多线程”

Node.js 默认是单线程事件循环模型。worker_threads 模块允许我们创建独立线程执行 JS 代码。

它的特点：

每个 Worker 有自己的事件循环和 V8 实例上下文
和主线程之间通过 消息传递 通信
适合 CPU 密集型场景
启动 Worker 有成本，不适合“一个任务一个 Worker”无限创建

所以生产里一般会做成 Worker Pool（线程池），复用线程，而不是每个任务都新建一个线程。

2. 消息队列的价值不只是“异步”

很多人第一次接触队列，理解停留在“把同步改成异步”。这没错，但不够。

消息队列更关键的价值在于：

削峰：请求量瞬间暴涨时，队列先兜住
解耦：生产者不关心消费者是否正忙
重试：失败任务可以重新投递
确认机制：只有处理完成才 ack
扩展性：消费者可横向扩容

3. 两者是如何配合的

消费链路可以理解成这样：

sequenceDiagram
    participant P as Producer
    participant Q as Queue
    participant C as Consumer
    participant W as Worker
    participant DB as ResultStore

    P->>Q: 发布任务
    Q->>C: 投递消息
    C->>W: 分发计算任务
    W-->>C: 返回结果/错误
    C->>DB: 更新任务状态
    C->>Q: ack / nack

关键点在于：

消费者不要自己做重计算
消费者只负责“调度”和“善后”
真正耗 CPU 的逻辑放进 Worker
ack 时机要慎重：通常在结果持久化成功后再 ack

4. 并发不是越高越好

一个很常见的误区是：
“我机器有 8 核，那我开 100 个 Worker，不就更快了吗？”

实际上会更慢。原因包括：

线程上下文切换开销
内存占用增加
CPU 抢占严重
GC 压力上升
队列积压时，过多并发会放大失败雪崩

经验上：

CPU 密集任务：Worker 数量一般从 CPU 核数 ~ 2 倍核数 试起
队列消费者的预取数（prefetch）要和 Worker 池容量匹配
用压测结果定，不要靠拍脑袋

方案对比与取舍分析

方案一：纯主线程异步

适合：

主要是 I/O 操作
任务执行非常轻
对延迟要求高但计算不重

不适合：

哈希计算、压缩、图像处理等 CPU 密集型任务

方案二：Child Process / Cluster

优点：

进程隔离强
某个任务崩了不一定拖垮主进程

缺点：

进程通信开销更大
内存占用通常高于 Worker Threads
管理复杂度更高

方案三：Worker Threads + 消息队列

优点：

能充分利用多核
线程通信成本较低
结合队列后具备更好的弹性和削峰能力

缺点：

需要设计线程池
需要处理任务状态、超时、重试、幂等
架构复杂度高于“单机脚本式处理”

一句话总结：
如果任务明显是 CPU 密集，并且有持续性高并发，Worker Threads + 消息队列通常是值得的。

容量估算：上线前别只看“能跑”

做架构时，我一般会先粗估 3 个值：

1. 到达速率

假设：

每秒进入系统 200 个任务

2. 单任务平均耗时

假设：

每个任务计算耗时 100ms

3. 理论处理能力

如果单个 Worker 每秒能处理约 10 个任务，那么：

8 个 Worker ≈ 每秒 80 个任务

这显然扛不住每秒 200 个任务的输入，结果就是：

队列积压持续增长
延迟越来越大

所以要么：

增加 Worker 数量和消费者实例
优化单任务计算耗时
对任务进行分级和限流
接受“异步延迟增长”的业务现实

一个简化估算公式：

所需并发处理能力 ≈ 输入 TPS × 单任务平均处理秒数

比如：

TPS = 200
平均耗时 = 0.1 秒

那么系统至少需要 20 的并发处理能力，且还要为波峰预留余量。

实战代码（可运行）

下面给一个简化但能跑的示例：

使用内存数组模拟消息队列
使用 Worker Threads 做 CPU 密集计算
使用线程池控制并发
用一个 API 提交任务并异步处理

说明：为了让示例更容易直接运行，这里不接 RabbitMQ/Kafka，而是先把结构讲明白。你理解这套代码后，把“内存队列”替换成真实 MQ 就很自然了。

目录结构

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

1）Worker 线程：执行真正的计算

task-worker.js

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

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

parentPort.on('message', (task) => {
  try {
    const { taskId, number } = task;
    const result = heavyCompute(number);
    parentPort.postMessage({
      taskId,
      success: true,
      result
    });
  } catch (error) {
    parentPort.postMessage({
      taskId,
      success: false,
      error: error.message
    });
  }
});

这里用“统计某个范围内质数个数”来模拟 CPU 密集计算。

2）线程池：复用 Worker

worker-pool.js

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

class WorkerPool {
  constructor(size = os.cpus().length) {
    this.size = size;
    this.workers = [];
    this.idleWorkers = [];
    this.taskQueue = [];
    this.callbacks = new Map();

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

  createWorker() {
    const worker = new Worker(path.resolve(__dirname, './task-worker.js'));

    worker.on('message', (message) => {
      const { taskId } = message;
      const callback = this.callbacks.get(taskId);

      if (callback) {
        this.callbacks.delete(taskId);
        callback.resolve(message);
      }

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

    worker.on('error', (err) => {
      console.error('Worker error:', err);
      this.replaceWorker(worker);
    });

    worker.on('exit', (code) => {
      if (code !== 0) {
        console.error(`Worker exited with code ${code}`);
        this.replaceWorker(worker);
      }
    });

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

  replaceWorker(worker) {
    this.workers = this.workers.filter((w) => w !== worker);
    this.idleWorkers = this.idleWorkers.filter((w) => w !== worker);
    this.createWorker();
    this.processNext();
  }

  runTask(task) {
    return new Promise((resolve, reject) => {
      this.taskQueue.push({ task, resolve, reject });
      this.processNext();
    });
  }

  processNext() {
    if (this.taskQueue.length === 0 || this.idleWorkers.length === 0) {
      return;
    }

    const worker = this.idleWorkers.pop();
    const { task, resolve, reject } = this.taskQueue.shift();

    this.callbacks.set(task.taskId, { resolve, reject });

    try {
      worker.postMessage(task);
    } catch (error) {
      this.callbacks.delete(task.taskId);
      this.idleWorkers.push(worker);
      reject(error);
    }
  }

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

module.exports = WorkerPool;

这个线程池很小巧，但已经体现了几个核心点：

固定大小线程池
空闲线程复用
排队等待
Worker 崩溃自动替换

3）主程序：模拟消息队列 + API 服务

app.js

const express = require('express');
const os = require('os');
const WorkerPool = require('./worker-pool');

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

const cpuCount = os.cpus().length;
const pool = new WorkerPool(Math.max(2, cpuCount - 1));

// 模拟任务队列
const messageQueue = [];
const taskStore = new Map();

function createTaskId() {
  return `${Date.now()}-${Math.random().toString(16).slice(2)}`;
}

// Producer: 接收任务
app.post('/tasks', (req, res) => {
  const { number } = req.body;

  if (!number || typeof number !== 'number' || number < 2) {
    return res.status(400).json({ error: 'number 必须是大于 1 的数字' });
  }

  const taskId = createTaskId();

  const task = {
    taskId,
    number,
    status: 'queued',
    createdAt: Date.now()
  };

  taskStore.set(taskId, task);
  messageQueue.push({ taskId, number });

  res.json({
    message: '任务已提交',
    taskId
  });
});

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

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

  res.json(task);
});

// Consumer: 从队列持续消费
async function consumeTasks() {
  while (true) {
    if (messageQueue.length === 0) {
      await sleep(100);
      continue;
    }

    const msg = messageQueue.shift();
    const task = taskStore.get(msg.taskId);

    if (!task) {
      continue;
    }

    task.status = 'processing';
    task.startedAt = Date.now();

    pool.runTask(msg)
      .then((result) => {
        task.finishedAt = Date.now();
        task.status = result.success ? 'done' : 'failed';
        task.result = result.result;
        task.error = result.error || null;
      })
      .catch((err) => {
        task.finishedAt = Date.now();
        task.status = 'failed';
        task.error = err.message;
      });
  }
}

function sleep(ms) {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

consumeTasks().catch(console.error);

const port = 3000;
app.listen(port, () => {
  console.log(`Server running at http://localhost:${port}`);
  console.log(`CPU count: ${cpuCount}, worker pool size: ${Math.max(2, cpuCount - 1)}`);
});

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

4）安装与运行

npm init -y
npm install express
node app.js

提交任务：

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

查询任务：

curl http://localhost:3000/tasks/你的taskId

架构演进：从示例到生产版怎么走

刚才的示例能说明机制，但离生产还有几步。

第一步：把内存队列换成真实消息队列

常见选择：

RabbitMQ：适合任务分发、ack/nack、死信队列
Kafka：适合高吞吐事件流，但“任务确认语义”需要额外设计
Redis Stream / BullMQ：Node.js 生态里上手快，适合中小型任务平台

如果你的目标是“明确的任务处理系统”，我一般更推荐：

RabbitMQ
或 Redis + BullMQ

因为它们在“任务重试、延时、失败处理”上更贴近业务需求。

第二步：增加任务状态机

任务不应该只有 queued/done/failed，最好有清晰状态流转：

stateDiagram-v2
    [*] --> queued
    queued --> processing
    processing --> done
    processing --> failed
    failed --> retrying
    retrying --> queued
    failed --> dead_letter
    done --> [*]
    dead_letter --> [*]

建议至少包含：

queued
processing
done
failed
retrying
dead_letter

第三步：结果持久化

内存 Map 只能演示，生产里必须落库，比如：

MySQL / PostgreSQL：查状态方便
Redis：适合短期状态缓存
对象存储：适合大型结果文件

第四步：加超时控制

有些任务不是“慢”，而是“卡死”。
如果一个 Worker 执行超过阈值，要能判定超时并处理。

常见坑与排查

这部分我想写得接地气一点，因为这些问题真的很常见，而且第一次遇到时会让人很懵。

坑 1：开了 Worker 还是感觉系统卡

现象

API 响应仍然抖动
CPU 使用率很高
任务处理速度没有明显提升

排查方向

是不是主线程还在做重逻辑
- 比如消息反序列化、数据预处理、结果聚合仍然很重
Worker 数量是不是开太多
- 超过 CPU 核数太多，线程抢占会让整体效率下降
是不是日志打太猛
- 高并发下大量 console.log 本身就是性能问题

建议

主线程只做路由、校验、调度
从 CPU 核数 - 1 左右开始压测
关掉无意义 debug 日志再看

坑 2：消息队列消费很快，但任务结果总是延迟很大

原因

你可能把“取消息”的并发开得很高，但 Worker 池容量有限。结果就是：

消费者拿到很多消息
实际只能排队等 Worker
队列表面上没积压，应用内存里却堆了一堆待处理任务

典型错误

MQ prefetch 设置成 500
Worker 池只有 8 个线程

建议

消费者并发 <= 实际可处理并发
prefetch 和 Worker 池大小联动配置
区分“MQ 积压”和“应用内部积压”

坑 3：Worker 崩了，任务凭空消失

原因

常见于 ack 时机不对：

刚拿到消息就 ack
然后 Worker 执行过程中挂了
任务实际上没完成，但队列认为已经处理成功

正确思路

任务结果持久化成功后再 ack
如果中途失败，nack 或重新入队
配合死信队列处理重试上限

坑 4：内存越来越高，最后 OOM

可能原因

待处理任务在内存堆积
Worker 消息传递对象太大
大对象未及时释放
返回结果过大，主线程缓存了太多

排查建议

看 process.memoryUsage()
看队列积压长度
控制单任务输入大小
避免在主线程保留完整大结果

如果任务传输的是大 Buffer，还要考虑：

是否能只传文件路径/对象存储地址
是否能分块处理而不是一次性塞进 Worker

坑 5：重复消费导致数据错乱

原因

消息队列通常只能保证“至少一次投递”，不能天然保证“恰好一次”。

这意味着：

任务可能被重复执行
尤其在消费者超时、网络抖动、进程重启时更常见

解决思路

做 幂等设计：

用任务唯一 ID 做去重
数据写入采用 upsert
对外部副作用操作做防重控制

安全/性能最佳实践

这一部分我尽量给“能落地”的建议，而不是泛泛而谈。

1. 任务入队前先做参数边界校验

不要让任意参数直接进入计算线程。

例如：

限制 number 的最大值
限制输入体积
校验任务类型白名单

否则有人传一个超大值，系统可能直接被拖死。

function validateTaskInput(number) {
  if (typeof number !== 'number' || Number.isNaN(number)) {
    throw new Error('number 必须是数字');
  }
  if (number < 2 || number > 1_000_000) {
    throw new Error('number 超出允许范围');
  }
}

2. 为任务设置超时

Worker 不要无限执行。

可以在主线程包一层超时控制：

function withTimeout(promise, ms) {
  return Promise.race([
    promise,
    new Promise((_, reject) =>
      setTimeout(() => reject(new Error('任务执行超时')), ms)
    )
  ]);
}

然后：

withTimeout(pool.runTask(msg), 5000)
  .then((result) => {
    // ...
  })
  .catch((err) => {
    // ...
  });

3. 控制队列长度和消费速率

如果入口无限收、后端处理有限，系统迟早出问题。

建议：

设置队列积压告警阈值
对 API 做限流
高峰期允许“降级接单”
为不同优先级任务分队列

4. 对大对象传输保持警惕

Worker 和主线程之间传消息，不是“零成本”的。
如果频繁传输大对象，会明显拖慢系统。

优先考虑：

传引用信息而不是传完整内容
只传任务 ID、文件路径、对象存储 key
必要时使用 Transferable 减少复制成本

5. 做好可观测性

至少监控这些指标：

队列长度
任务成功率 / 失败率
平均处理时长 / P95 / P99
Worker 池繁忙度
进程 CPU / 内存
重试次数
死信队列数量

没有监控的高并发系统，出了问题基本只能靠猜。

6. 任务要有重试上限

不要无限重试。
否则某类坏任务会一直占资源。

推荐做法：

可重试错误：设置 3~5 次上限 + 指数退避
不可重试错误：直接失败
超过阈值进入死信队列，人工排查

一个更接近生产的落地建议

如果你准备真正上线，我建议按下面的节奏推进：

小规模阶段

适合：

日任务量不大
团队希望快速验证

方案：

Node.js API
BullMQ / RabbitMQ
Worker Threads 线程池
Redis / MySQL 存任务状态

中等规模阶段

适合：

有明显高峰低谷
要求失败可追踪、延迟可观测

增强点：

死信队列
指数退避重试
任务优先级
实例水平扩容
指标监控和告警

更大规模阶段

适合：

多种任务类型
资源消耗差异明显
需要弹性调度