Node.js cluster模块

cluster 模块实现多进程架构，突破单线程限制。

核心概念

架构模型

判断进程类型

const cluster = require('cluster');

if (cluster.isMaster || cluster.isPrimary) {
  console.log('主进程 PID:', process.pid);
} else {
  console.log('工作进程 PID:', process.pid);
}

基本用法

创建集群

const cluster = require('cluster');
const http = require('http');
const numCPUs = require('os').cpus().length;

if (cluster.isMaster) {
  console.log(`主进程 ${process.pid} 启动`);

  // 创建工作进程
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  cluster.on('exit', (worker, code, signal) => {
    console.log(`工作进程 ${worker.process.pid} 退出`);
    cluster.fork(); // 自动重启
  });
} else {
  http.createServer((req, res) => {
    res.writeHead(200);
    res.end(`进程 ${process.pid} 响应\n`);
  }).listen(8000);

  console.log(`工作进程 ${process.pid} 启动`);
}

负载均衡策略

Round-Robin（默认）

// Master 按轮询顺序分发请求
// Linux/Windows 默认启用

cluster.schedulingPolicy = cluster.SCHED_RR; // 显式设置

共享端口

// 所有 Worker 共享同一端口
// Master 接收请求后分发到 Worker

if (cluster.isMaster) {
  cluster.fork();
  cluster.fork();
} else {
  // Worker 监听同一端口
  http.createServer().listen(8000);
}

进程管理

获取 Worker 信息

if (cluster.isMaster) {
  // 所有工作进程
  console.log(cluster.workers);

  // 遍历
  Object.values(cluster.workers).forEach((worker) => {
    console.log(`Worker ${worker.id}: PID ${worker.process.pid}`);
  });
}

Worker 对象

const worker = cluster.fork();

// 属性
worker.id;          // Worker ID
worker.process;     // 进程对象
worker.process.pid; // 进程 PID

// 方法
worker.send(msg);       // 发送消息
worker.disconnect();    // 断开连接
worker.kill();          // 终止进程
worker.kill('SIGTERM'); // 发送信号
worker.isDead();        // 是否已终止
worker.isConnected();   // 是否连接中

事件监听

Master 事件

cluster.on('fork', (worker) => {
  console.log(`Worker ${worker.id} forked`);
});

cluster.on('online', (worker) => {
  console.log(`Worker ${worker.id} online`);
});

cluster.on('listening', (worker, address) => {
  console.log(`Worker ${worker.id} listening ${address.port}`);
});

cluster.on('disconnect', (worker) => {
  console.log(`Worker ${worker.id} disconnected`);
});

cluster.on('exit', (worker, code, signal) => {
  console.log(`Worker ${worker.id} exit, code: ${code}`);
});

cluster.on('message', (worker, msg) => {
  console.log(`Worker ${worker.id} 发送:`, msg);
});

Worker 事件

// 在 Worker 进程中
process.on('message', (msg) => {
  console.log('收到 Master 消息:', msg);
});

进程间通信

Master 发送

if (cluster.isMaster) {
  const worker = cluster.fork();

  worker.send({ type: 'config', data: 'settings' });

  worker.on('message', (msg) => {
    console.log('Worker 响应:', msg);
  });
}

Worker 发送

// worker.js
process.on('message', (msg) => {
  if (msg.type === 'config') {
    process.send({ type: 'ready' });
  }
});

广播消息

if (cluster.isMaster) {
  // 向所有 Worker 发送
  Object.values(cluster.workers).forEach((worker) => {
    worker.send({ type: 'broadcast' });
  });
}

自动重启

if (cluster.isMaster) {
  const numCPUs = require('os').cpus().length;

  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  cluster.on('exit', (worker, code, signal) => {
    if (code !== 0) {
      console.log(`Worker ${worker.id} 异常退出，重启中...`);
      cluster.fork();
    }
  });
}

环境变量传递

if (cluster.isMaster) {
  // fork 时设置环境变量
  cluster.fork({ PORT: 8001 });
  cluster.fork({ PORT: 8002 });
}

// Worker 中获取
console.log(process.env.PORT);

注意事项

• Worker 数量建议等于 CPU 核心数
• 共享内存需使用 Redis 等外部存储
• Worker 崩溃后应重启保证稳定
• Windows 默认使用 SCHED_NONE（随机分发）

要点总结

• cluster.fork() 创建工作进程
• 所有 Worker 共享同一端口
• 默认 Round-Robin 负载均衡
• cluster.on('exit') 监听崩溃并重启
• worker.send() 和 process.on('message') 通信