如何让Node.js集群跳过繁忙的Worker进程？

我正在使用Node.js 18编写Web服务器，其中/slow路由为计算密集型任务，会长时间运行并将CPU核心占满至100%。为保证其他请求响应性，我采用了Node.js cluster模块。

简化后的代码 #

import cluster from 'cluster';
import { createServer } from 'http';

const startUpTime = Date.now();

function requestListener(request, response) {
  console.log(`[${(Date.now()-startUpTime)/1000}] pid ${process.pid} is serving URL ${request.url}`);
  response.writeHead(200);
  if (request.url === '/slow') {
    for (let cnt = 0; cnt < 100; cnt++) {
      for (let i = 0; i < 1000000000; i++ ) { /* just wait */ }
      response.write(`${cnt}.`);
    }
  }
  response.end(`... pid ${process.pid} done...\n`);
  console.log(`[${(Date.now()-startUpTime)/1000}] pid ${process.pid} finished serving URL ${request.url}`);
}

if (cluster.isPrimary) {
  for (let i = 0; i < 4; i++) {
    let worker = cluster.fork();
  }
} else {
  // the worker:
  const server = createServer(requestListener);
  server.listen(8000, 'localhost', () => {
    console.info(`Server ${process.pid} is running`);
  });
}

问题现象 #

系统有12个核心且基本空闲，发起多个/test请求后，第7个/test请求被分配给正在处理/slow的繁忙Worker进程（pid 78798），导致请求被阻塞，此时还有3个空闲Worker进程。

控制台输出：

Server 78797 is running
Server 78799 is running
Server 78798 is running
Server 78800 is running
[3.028] pid 78797 is serving URL /test
[3.029] pid 78797 finished serving URL /test
[3.515] pid 78799 is serving URL /test
[3.52] pid 78799 finished serving URL /test
[5.963] pid 78798 is serving URL /slow
[7.27] pid 78800 is serving URL /test
[7.272] pid 78800 finished serving URL /test
[7.976] pid 78797 is serving URL /test
[7.976] pid 78797 finished serving URL /test
[8.568] pid 78799 is serving URL /test
[8.569] pid 78799 finished serving URL /test
[32.492] pid 78798 finished serving URL /slow
[32.495] pid 78798 is serving URL /test
[32.495] pid 78798 finished serving URL /test

根据Node.js文档，默认轮询调度具备避免Worker过载的智能特性，但实际未生效。请问如何修改该行为，确保所有请求都能立即响应？

问题原因 #

Node.js默认的轮询调度（cluster.SCHED_RR）基于文件描述符分发请求，当Worker被CPU密集型任务完全占用时，无法及时向主进程发送"就绪"信号，主进程会误判该Worker仍可处理请求，继续分发新请求到该Worker。

解决方法 #

1. 切换到操作系统级调度策略（最简单有效） #

将cluster的调度策略改为cluster.SCHED_NONE，让操作系统负责连接分发。操作系统能更精准识别空闲Worker，自动将新请求分配给未被占用的进程。

修改主进程代码，在fork Worker前添加：

if (cluster.isPrimary) {
  // 设置调度策略为操作系统负责
  cluster.schedulingPolicy = cluster.SCHED_NONE;
  
  for (let i = 0; i < 4; i++) {
    let worker = cluster.fork();
  }
}

2. 手动实现负载均衡（更灵活） #

主进程自己监听端口，维护Worker的负载状态（如当前处理请求数），主动将请求转发给空闲Worker。这种方式适合需要自定义负载逻辑的场景。

示例修改后的完整代码：

import cluster from 'cluster';
import { createServer } from 'http';

const startUpTime = Date.now();
// 主进程维护Worker的空闲状态
const workerStatus = new Map();

function requestListener(request, response) {
  console.log(`[${(Date.now()-startUpTime)/1000}] pid ${process.pid} is serving URL ${request.url}`);
  response.writeHead(200);
  if (request.url === '/slow') {
    for (let cnt = 0; cnt < 100; cnt++) {
      for (let i = 0; i < 1000000000; i++ ) { /* just wait */ }
      response.write(`${cnt}.`);
    }
  }
  response.end(`... pid ${process.pid} done...\n`);
  console.log(`[${(Date.now()-startUpTime)/1000}] pid ${process.pid} finished serving URL ${request.url}`);
  // 通知主进程当前Worker已空闲
  process.send({ type: 'free' });
}

if (cluster.isPrimary) {
  for (let i = 0; i < 4; i++) {
    const worker = cluster.fork();
    workerStatus.set(worker.process.pid, { isBusy: false, worker });
    // 监听Worker的状态消息
    worker.on('message', (msg) => {
      if (msg.type === 'free') {
        workerStatus.set(worker.process.pid, { isBusy: false, worker });
      }
    });
  }

  // 主进程监听端口，手动分发请求
  const server = createServer((req, res) => {
    // 查找第一个空闲的Worker
    let freeWorker = null;
    for (const [pid, status] of workerStatus.entries()) {
      if (!status.isBusy) {
        freeWorker = status.worker;
        break;
      }
    }

    if (freeWorker) {
      workerStatus.set(freeWorker.process.pid, { isBusy: true, worker: freeWorker });
      // 将请求转发给Worker
      freeWorker.send({ type: 'request', req, res });
    } else {
      // 所有Worker繁忙时返回服务不可用
      res.writeHead(503);
      res.end('All workers are busy');
    }
  });

  server.listen(8000, 'localhost', () => {
    console.info(`Primary server ${process.pid} is running`);
  });
} else {
  // Worker监听主进程的请求消息
  process.on('message', (msg) => {
    if (msg.type === 'request') {
      requestListener(msg.req, msg.res);
    }
  });
}

3. 结合Worker Threads处理密集任务 #

对于/slow这类计算密集型任务，可在单个Worker内部使用worker_threads模块开启子线程处理，避免整个Worker被阻塞。这种方式适合任务可拆分的场景，配合cluster横向扩展能进一步提升并发能力。

内容的提问来源于stack exchange，提问作者Chris