背景与问题

很多团队在用 Node.js 做服务时，前期都很顺：I/O 密集型接口吞吐高、开发效率快、部署简单。但一旦业务里混入CPU 密集型任务，比如图片处理、报表计算、批量加解密、规则引擎、向量计算，问题就会开始暴露：

• 接口 RT 突然飙高
• 事件循环被阻塞，健康检查都开始超时
• 单机 CPU 明明很多核，但 Node 进程只忙一个核
• 高峰期任务大量堆积，内存上涨，甚至 OOM
• 简单 Promise.all() 看起来很“并发”，实际上只是把主线程压得更惨

我第一次遇到这个问题，是在一个“批量生成报表”的场景里。最开始我们直接在 HTTP 请求里做计算，测试环境没问题，一上生产，高峰时所有接口都被拖慢。后来才意识到：Node.js 的强项是异步 I/O，不是把 CPU 活硬塞进主线程。

这类问题通常不能只靠“再开几个实例”解决。更稳妥的办法往往是两层拆分：

1. 消息队列：把请求流量和任务执行解耦，削峰填谷；
2. Worker Threads：把 CPU 密集型任务从主线程搬走，真正利用多核。

这篇文章我会从架构角度，带你把这两者组合成一套可运行、可扩展、可排障的高并发任务处理方案。

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

如果只用 Worker Threads，不用消息队列，会怎样？

• 好处：同一个进程内就能并行计算，开发简单；
• 问题：任务来源还是瞬时直冲应用进程，高峰期容易把内存打爆；
• 问题：服务重启、进程崩溃时，任务状态不好恢复；
• 问题：缺少天然的重试、积压、延迟、削峰能力。

如果只用消息队列，不用 Worker Threads，又会怎样？

• 好处：任务异步化了，接口不阻塞；
• 问题：消费者如果还是主线程执行 CPU 活，照样卡事件循环；
• 问题：单消费者吞吐受限，单进程无法充分吃满多核。

所以更合理的组合是：

• 消息队列负责“排队、缓冲、重试、解耦”
• Worker Threads 负责“多核并行执行 CPU 密集型任务”

这套思路特别适合下面这些场景：

• 批量文件处理
• 图像/音视频转码前后处理
• 风控规则批量计算
• 大量 PDF/报表生成
• 批量数据清洗、压缩、加密
• AI 前后处理中的文本切分、向量预处理等

核心原理

1. Node.js 事件循环不适合长时间 CPU 计算

Node.js 的主线程负责事件循环。I/O 操作适合放在这里，因为大部分时间都在等待。但如果你执行的是大循环、复杂计算、同步压缩、同步 JSON 大对象处理，这些都会让事件循环长时间得不到释放。

一个简单判断标准：

• I/O 密集型：适合主线程异步处理
• CPU 密集型：适合 Worker Threads 或独立进程

2. Worker Threads 的定位

worker_threads 是 Node.js 官方提供的多线程能力。它不是为了替代所有异步模型，而是为了把重计算从主线程拆出去。

它的核心特征：

• 每个 Worker 有自己的 JS 运行环境和事件循环
• 可以通过 postMessage 通信
• 可以共享 SharedArrayBuffer
• 适合做线程池，而不是每个任务临时创建线程

临时创建 Worker 的开销不低，所以实践里通常会做一个固定大小的线程池。

3. 消息队列的定位

消息队列的价值不只是“异步”，更关键是：

• 削峰：把瞬时流量变成可控消费速率
• 缓冲：消费者忙不过来时，任务先排队
• 可靠性：失败可重试，崩溃后任务不丢
• 解耦：API 层、调度层、执行层职责更清晰

本文实战代码为了方便运行，我会用一个内存队列模拟消息队列。真实生产环境建议替换为：

• Redis + BullMQ
• RabbitMQ
• Kafka
• 云厂商托管消息服务

4. 推荐处理链路

flowchart LR
    A[客户端请求] --> B[API 服务]
    B --> C[写入消息队列]
    C --> D[消费者调度器]
    D --> E[Worker 线程池]
    E --> F[任务执行]
    F --> G[结果存储/回调/状态更新]

这个链路有个关键思想：请求接入与任务执行解耦。
HTTP 请求只负责“创建任务”，真正执行交给后面的消费者和线程池。

架构设计与取舍分析

分层职责

为了避免代码一团糟，建议把系统拆成 4 层：

1. 接入层

   • 接 HTTP 请求
   • 做鉴权、参数校验、限流
   • 写入任务队列

2. 调度层

   • 从消息队列取任务
   • 控制消费速率、并发数、重试
   • 将任务分派到 Worker 池

3. 执行层

   • 真正跑 CPU 密集型任务
   • 尽量无状态
   • 能超时、能失败、能上报错误

4. 结果层

   • 更新任务状态
   • 存储结果
   • 触发回调或通知

与常见方案对比

容量估算思路

做架构设计时，别只看“能不能跑”，还要大概算一下容量。

一个粗略公式：

单机任务吞吐 ≈ Worker 数 × 每个任务平均处理速度

例如：

• 8 核机器
• Worker 池大小设为 6
• 平均每个任务 200ms

那么理论吞吐大约：

6 / 0.2 = 30 个任务/秒

再考虑上下文切换、消息传递、序列化开销，真实值可能打个 0.6 ~ 0.8 折。

如果入队速度长期高于消费速度，就会出现积压。
因此你至少需要监控：

• 队列长度
• 平均等待时间
• 任务执行时间
• 重试率
• Worker 忙碌率

核心流程图

下面这张图展示了一个完整的任务生命周期：

sequenceDiagram
    participant Client as 客户端
    participant API as API服务
    participant MQ as 消息队列
    participant Scheduler as 调度器
    participant Pool as Worker池
    participant Worker as Worker线程
    participant Store as 状态存储

    Client->>API: 提交任务
    API->>MQ: 入队
    API->>Store: 记录任务状态=queued
    Scheduler->>MQ: 拉取任务
    Scheduler->>Pool: 分配可用Worker
    Pool->>Worker: postMessage(task)
    Worker->>Worker: 执行CPU密集型计算
    Worker-->>Pool: 返回结果/错误
    Pool->>Store: 更新状态=done/failed
    Scheduler->>MQ: ack / retry

实战代码（可运行）

下面我们做一个可直接运行的示例，演示：

• 用 Express 接收任务
• 用内存队列模拟消息队列
• 用 Worker Threads 线程池执行 CPU 密集型任务
• 支持任务状态查询
• 支持失败重试和超时控制

为了让示例聚焦核心原理，我这里用“计算斐波那契数”模拟 CPU 密集型任务。真实项目里，你可以替换成报表、图片处理、加密等业务逻辑。

目录结构

node-worker-queue-demo/
├─ package.json
├─ server.js
├─ queue.js
├─ worker-pool.js
└─ worker.js

package.json

{
  "name": "node-worker-queue-demo",
  "version": "1.0.0",
  "type": "commonjs",
  "main": "server.js",
  "scripts": {
    "start": "node server.js"
  },
  "dependencies": {
    "express": "^4.21.1"
  }
}

安装依赖：

npm install
npm start

worker.js

Worker 中执行真正的 CPU 密集型任务。

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

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

parentPort.on('message', async (job) => {
  const startedAt = Date.now();

  try {
    const { jobId, payload } = job;
    const { n } = payload;

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

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

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

这里故意限制 n <= 40，是为了避免恶意请求把 CPU 直接打满。这是生产实践里非常重要的一点，后面还会展开讲。

worker-pool.js

实现一个最小可用线程池。

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

class WorkerPool {
  constructor(size = 4, timeout = 10000) {
    this.size = size;
    this.timeout = timeout;
    this.workers = [];
    this.idleWorkers = [];
    this.taskQueue = [];
    this.runningTasks = new Map();
  }

  init() {
    for (let i = 0; i < this.size; i++) {
      const worker = new Worker(path.resolve(__dirname, './worker.js'));

      worker.on('message', (message) => {
        const currentTask = this.runningTasks.get(worker);
        if (!currentTask) return;

        clearTimeout(currentTask.timer);
        this.runningTasks.delete(worker);

        if (message.ok) {
          currentTask.resolve(message);
        } else {
          currentTask.reject(new Error(message.error));
        }

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

      worker.on('error', (err) => {
        const currentTask = this.runningTasks.get(worker);
        if (currentTask) {
          clearTimeout(currentTask.timer);
          currentTask.reject(err);
          this.runningTasks.delete(worker);
        }

        this._replaceWorker(worker);
        this._drain();
      });

      worker.on('exit', (code) => {
        if (code !== 0) {
          const currentTask = this.runningTasks.get(worker);
          if (currentTask) {
            clearTimeout(currentTask.timer);
            currentTask.reject(new Error(`Worker exited with code ${code}`));
            this.runningTasks.delete(worker);
          }

          this._replaceWorker(worker);
          this._drain();
        }
      });

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

  exec(job) {
    return new Promise((resolve, reject) => {
      this.taskQueue.push({ job, resolve, reject });
      this._drain();
    });
  }

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

      const timer = setTimeout(() => {
        this.runningTasks.delete(worker);
        task.reject(new Error('任务执行超时'));
        this._replaceWorker(worker);
        this._drain();
      }, this.timeout);

      this.runningTasks.set(worker, {
        ...task,
        timer
      });

      worker.postMessage(task.job);
    }
  }

  _replaceWorker(oldWorker) {
    this.workers = this.workers.filter((w) => w !== oldWorker);
    this.idleWorkers = this.idleWorkers.filter((w) => w !== oldWorker);
    this.runningTasks.delete(oldWorker);

    try {
      oldWorker.terminate();
    } catch (_) {}

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

    worker.on('message', (message) => {
      const currentTask = this.runningTasks.get(worker);
      if (!currentTask) return;

      clearTimeout(currentTask.timer);
      this.runningTasks.delete(worker);

      if (message.ok) {
        currentTask.resolve(message);
      } else {
        currentTask.reject(new Error(message.error));
      }

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

    worker.on('error', (err) => {
      const currentTask = this.runningTasks.get(worker);
      if (currentTask) {
        clearTimeout(currentTask.timer);
        currentTask.reject(err);
        this.runningTasks.delete(worker);
      }
      this._replaceWorker(worker);
    });

    worker.on('exit', (code) => {
      if (code !== 0) {
        const currentTask = this.runningTasks.get(worker);
        if (currentTask) {
          clearTimeout(currentTask.timer);
          currentTask.reject(new Error(`Worker exited with code ${code}`));
          this.runningTasks.delete(worker);
        }
        this._replaceWorker(worker);
      }
    });

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

  stats() {
    return {
      size: this.size,
      idleWorkers: this.idleWorkers.length,
      busyWorkers: this.runningTasks.size,
      waitingTasks: this.taskQueue.length
    };
  }
}

module.exports = WorkerPool;

queue.js

用一个简单的内存队列模拟消息队列，支持重试。

class InMemoryQueue {
  constructor() {
    this.queue = [];
  }

  push(job) {
    this.queue.push(job);
  }

  pop() {
    return this.queue.shift();
  }

  size() {
    return this.queue.length;
  }
}

module.exports = InMemoryQueue;

server.js

整合 API、队列、调度器、线程池。

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

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

const queue = new InMemoryQueue();
const jobStore = new Map();

const cpuCount = os.cpus().length;
const poolSize = Math.max(1, Math.min(cpuCount - 1, 6));
const pool = new WorkerPool(poolSize, 8000);
pool.init();

const MAX_RETRY = 2;
const SCHEDULER_INTERVAL = 50;
let schedulerRunning = false;

function createJob(payload) {
  return {
    jobId: crypto.randomUUID(),
    payload,
    retryCount: 0,
    createdAt: Date.now()
  };
}

async function processJob(job) {
  jobStore.set(job.jobId, {
    status: 'running',
    payload: job.payload,
    retryCount: job.retryCount,
    updatedAt: Date.now()
  });

  const result = await pool.exec(job);

  jobStore.set(job.jobId, {
    status: 'done',
    payload: job.payload,
    result: result.result,
    duration: result.duration,
    retryCount: job.retryCount,
    updatedAt: Date.now()
  });
}

async function schedulerLoop() {
  if (schedulerRunning) return;
  schedulerRunning = true;

  while (true) {
    const poolStats = pool.stats();

    if (poolStats.idleWorkers > 0 && queue.size() > 0) {
      const job = queue.pop();

      try {
        await processJob(job);
      } catch (error) {
        if (job.retryCount < MAX_RETRY) {
          job.retryCount += 1;
          jobStore.set(job.jobId, {
            status: 'retrying',
            payload: job.payload,
            retryCount: job.retryCount,
            error: error.message,
            updatedAt: Date.now()
          });
          queue.push(job);
        } else {
          jobStore.set(job.jobId, {
            status: 'failed',
            payload: job.payload,
            retryCount: job.retryCount,
            error: error.message,
            updatedAt: Date.now()
          });
        }
      }
    } else {
      await new Promise((resolve) => setTimeout(resolve, SCHEDULER_INTERVAL));
    }
  }
}

app.post('/jobs', (req, res) => {
  const { n } = req.body || {};

  if (!Number.isInteger(n)) {
    return res.status(400).json({
      error: '参数 n 必须是整数'
    });
  }

  const job = createJob({ n });

  jobStore.set(job.jobId, {
    status: 'queued',
    payload: job.payload,
    retryCount: 0,
    updatedAt: Date.now()
  });

  queue.push(job);

  res.status(202).json({
    message: '任务已入队',
    jobId: job.jobId
  });
});

app.get('/jobs/:jobId', (req, res) => {
  const data = jobStore.get(req.params.jobId);

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

  res.json({
    jobId: req.params.jobId,
    ...data
  });
});

app.get('/metrics', (req, res) => {
  res.json({
    queueSize: queue.size(),
    pool: pool.stats(),
    totalJobs: jobStore.size
  });
});

const port = 3000;
app.listen(port, () => {
  console.log(`Server started at http://localhost:${port}`);
  console.log(`CPU cores: ${cpuCount}, worker pool size: ${poolSize}`);
  schedulerLoop().catch((err) => {
    console.error('Scheduler crashed:', err);
    process.exit(1);
  });
});

运行与验证

1）提交任务

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

返回示例：

{
  "message": "任务已入队",
  "jobId": "2d0f0da0-8e4d-4e8e-9a4f-d91f53b8c6d0"
}

2）查询任务状态

curl http://localhost:3000/jobs/2d0f0da0-8e4d-4e8e-9a4f-d91f53b8c6d0

可能看到：

{
  "jobId": "2d0f0da0-8e4d-4e8e-9a4f-d91f53b8c6d0",
  "status": "done",
  "payload": {
    "n": 35
  },
  "result": 9227465,
  "duration": 168,
  "retryCount": 0,
  "updatedAt": 1731646690000
}

3）查看队列与线程池指标

curl http://localhost:3000/metrics

关键实现细节解读

上面的示例能跑，但理解“为什么这么写”更重要。

1. 为什么要线程池，而不是每个任务创建一个 Worker

因为 Worker 创建本身有成本：

• 初始化运行时
• 加载模块
• 建立通信通道

如果每来一个任务就起一个 Worker，在高并发下会产生明显抖动，甚至线程数失控。
所以更合理的是：

• 固定池大小
• 任务排队
• Worker 复用

2. 为什么调度器要控制消费，而不是一把梭哈全取出来

很多初学者做队列消费时，会下意识写成：

while(queue.length) {
  const job = queue.shift();
  pool.exec(job);
}

看起来很“快”，实际上会导致：

• 大量任务同时进入应用内存
• 下游执行能力不够时形成应用内堆积
• 失败重试风暴更难控制

真正稳的方式是：按可用 Worker 数量进行有界调度。

3. 为什么要设置超时

CPU 任务也会卡死，常见原因包括：

• 输入异常导致极端计算
• 第三方 native 模块挂住
• 代码死循环
• 共享内存竞争问题

超时策略至少能做到两件事：

• 避免单个任务长期占住 Worker
• 为自动恢复创造条件

再进一步：更贴近生产的架构

上面的示例为了便于理解，使用的是单进程内存队列。生产环境更推荐下面这种结构：

flowchart TB
    A[API 服务实例1] --> MQ[(消息队列)]
    B[API 服务实例2] --> MQ
    C[API 服务实例3] --> MQ

    MQ --> D[消费者实例1]
    MQ --> E[消费者实例2]

    D --> F[Worker池]
    E --> G[Worker池]

    F --> H[(任务结果存储)]
    G --> H

这种架构的优势：

• API 实例可以横向扩容
• 消费者实例可以独立扩容
• 某个消费者挂掉，不影响队列中的任务
• 任务执行层与接入层彻底解耦

如果你用 Redis 生态，常见选择是：

• BullMQ 做队列管理
• Worker Threads 作为消费者进程内部的并行执行层

这样组合下来，落地成本通常比较平衡。

常见坑与排查

这一部分我想讲得稍微“接地气”一些，因为这些坑真的是很容易踩。

1. 以为 async/await 能解决 CPU 阻塞

这是最常见误区之一。
async/await 只能让异步流程写起来像同步，但不会把 CPU 计算自动变成多线程。

现象

• 接口用了 await
• 但 CPU 一高，整个服务还是慢

排查方法

• 看 Node 进程 CPU 是否长期打满
• 用 clinic doctor、0x 或 CPU profile 分析热点
• 观察事件循环延迟

结论

只要是纯 JS 重计算，还是得上 Worker 或多进程。

2. 任务对象太大，线程通信成本高

Worker 和主线程之间传递消息需要序列化/拷贝，数据太大会带来明显开销。

现象

• Worker 明明不忙，但整体吞吐上不去
• 内存波动大
• GC 频繁

排查方法

• 统计单个任务 payload 大小
• 比较“传大对象”和“传文件路径/对象 ID”的差异

建议

• 不要直接传大 Buffer、大数组、大 JSON
• 更推荐传引用信息，比如文件路径、对象存储 key、数据库记录 ID
• 极端性能要求下考虑 Transferable 或 SharedArrayBuffer

3. Worker 数开太多，反而更慢

不少人看到 16 核机器，就直接开 16、32 甚至更多 Worker。结果吞吐没有提高，延迟还更差。

原因

• 上下文切换增加
• CPU cache 命中率下降
• 主线程调度开销增加
• 内存占用放大

经验值

• 纯 CPU 任务：通常从 CPU 核数 - 1 或 CPU 核数 * 0.5 ~ 1 开始试
• 不要靠拍脑袋，压测才有答案

4. 队列重试没有幂等，导致重复执行

消息队列天然可能出现至少一次投递。如果任务失败重试，但业务不幂等，就会产生重复结果。

典型场景

• 重复扣费
• 重复发送通知
• 重复写入一批统计数据

建议

• 每个任务有唯一业务 ID
• 消费端做幂等校验
• 结果写入使用去重约束或状态机控制

5. 只监控接口 RT，不监控队列积压

这也是我见过非常多的盲点。
接口返回 202 很快，看起来一切正常，但后面的队列已经堆了几十万条。

必监控指标

• 队列长度
• 最老任务等待时长
• 单任务执行耗时 P95/P99
• Worker 忙碌率
• 重试率/失败率
• 进程 RSS / heapUsed

6. Worker 崩了但系统没有自愈

如果 Worker 执行中抛出致命错误、native 模块崩溃、线程异常退出，而线程池不补位，系统吞吐会慢慢掉下去。

建议

• 监听 error 和 exit
• 异常退出后自动补新 Worker
• 对长期异常的任务类型做熔断或降级

安全/性能最佳实践

这一节我会把“能上线”的关键点集中列出来。

安全实践

1. 严格限制任务输入

对于 CPU 密集型任务，输入规模往往决定资源消耗上限。
所以一定要限制：

• 数值范围
• 数组长度
• 文本大小
• 文件大小
• 嵌套层级

例如：

if (n > 40) {
  throw new Error('n too large');
}

这不是偷懒，而是防止恶意或误操作把机器打爆。

2. API 层做鉴权与限流

消息队列不是无限垃圾桶。
如果接入层没有限流，攻击者可以快速制造海量任务，队列会被塞爆。

建议：

• 用户级限流
• IP 限流
• 任务类型配额
• 总队列上限保护

3. 不要在 Worker 中执行不可信代码

Worker 不是安全沙箱。
如果你有“用户提交脚本执行”这类需求，不要直接扔进 Worker。应考虑：

• 独立容器
• seccomp / namespace 等系统级隔离
• 沙箱服务

性能实践

1. 任务拆分粒度要合适

太细：

• 消息传递成本高
• 调度开销高

太粗：

• 单任务耗时长
• 重试成本高
• 长尾严重

经验上，单任务执行时间落在几十毫秒到几百毫秒，通常比较容易调优。

2. 避免主线程做大对象序列化

主线程的职责是“快进快出”：

• 接任务
• 校验参数
• 更新状态
• 分发执行

如果主线程还负责组装超大对象、做大 JSON stringify/parse，就会影响整体吞吐。

3. 给队列设置背压机制

背压的本质是：下游处理不过来时，上游必须收敛。

常见做法：

• 队列长度超过阈值时拒绝新任务
• 对低优先级任务降级
• 先返回“系统繁忙，请稍后再试”
• 动态调整消费者并发

4. 结果尽量外部存储，不要全堆内存

示例里用 Map 存状态，是为了简单。生产环境里应改成：

• Redis：适合状态缓存、短期结果
• MySQL/PostgreSQL：适合任务审计和长期记录
• 对象存储：适合大结果文件

5. 做好优先级和隔离

不是所有任务都应该进同一个池子。

例如：

• 短任务池：实时性要求高
• 长任务池：吞吐优先
• 高优任务队列：核心业务
• 低优任务队列：离线补偿

否则一个大批量低优任务就可能把高优任务全部堵死。

一个更稳妥的生产落地建议

如果你准备把这套模式真正用到线上，我建议按下面顺序演进：

阶段一：单体服务内引入 Worker 池

适合场景：

• 任务量不大
• 先解决主线程阻塞
• 快速验证效果

特点：

• 改造成本低
• 见效快
• 但可靠性和扩展性一般

阶段二：引入 Redis/BullMQ 等消息队列

适合场景：

• 任务异步化需求明确
• 有失败重试、延迟重试、积压管理需求

特点：

• 可靠性更高
• 可观察性更好
• 开始具备削峰能力

阶段三：消费者服务独立部署，内部使用 Worker 池

适合场景：

• 任务量持续增长
• API 与执行层需要独立扩容
• 高峰明显、任务种类变多

特点：

• 架构更清晰
• 易做优先级隔离
• 适合中大型系统

适用边界与不适用场景

这个方案很好，但也不是万能钥匙。

适合

• CPU 密集型任务明显
• 任务可异步化
• 允许最终一致
• 有高峰流量、需要削峰

不太适合

• 强实时、必须同步返回结果的极短请求链路
• 任务强依赖共享内存复杂状态
• 任务本身更适合 GPU 或专用计算服务
• 需要极强安全隔离的不可信执行环境

如果任务是纯 I/O，比如调多个 HTTP 接口、查数据库、读对象存储，通常没必要上 Worker Threads，先把异步 I/O 和连接池调优做好更重要。

总结

把 Node.js 用好，关键不是让它“什么都干”，而是让它干自己擅长的事：

• 主线程负责接入、调度、状态流转
• 消息队列负责削峰、缓冲、重试、解耦
• Worker Threads负责 CPU 密集型并行计算

这套“消息队列 + Worker 池”的架构，本质上是在解决三个问题：

1. 主线程不能被 CPU 任务拖死
2. 高峰流量不能直接压垮执行层
3. 任务失败、重试、积压必须可控可观测

如果你现在正面临 Node.js 高并发任务处理瓶颈，我建议按这个顺序落地：

1. 先识别 CPU 密集型热点，别盲目优化所有接口
2. 把重计算迁出主线程，先用 Worker 池验证收益
3. 再引入真正的消息队列，解决削峰与可靠性
4. 补齐超时、重试、幂等、监控、背压
5. 最后通过压测决定 Worker 数、消费者数和队列阈值

一句话总结：
Worker Threads 解决“算得动”，消息队列解决“扛得住”，两者结合才能让 Node.js 在高并发任务场景下跑得稳。