Cluster (集群)#

稳定性:2 - 稳定

源代码: lib/cluster.js

Node.js 进程集群可用于运行多个 Node.js 实例,这些实例可以在其应用程序线程之间分配工作负载。当不需要进程隔离时,请改用 worker_threads 模块,它允许在单个 Node.js 实例中运行多个应用程序线程。

cluster 模块可以轻松创建共享服务器端口的子进程。

import cluster from 'node:cluster';
import http from 'node:http';
import { availableParallelism } from 'node:os';
import process from 'node:process';

const numCPUs = availableParallelism();

if (cluster.isPrimary) {
  console.log(`Primary ${process.pid} is running`);

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

  cluster.on('exit', (worker, code, signal) => {
    console.log(`worker ${worker.process.pid} died`);
  });
} else {
  // Workers can share any TCP connection
  // In this case it is an HTTP server
  http.createServer((req, res) => {
    res.writeHead(200);
    res.end('hello world\n');
  }).listen(8000);

  console.log(`Worker ${process.pid} started`);
}const cluster = require('node:cluster');
const http = require('node:http');
const numCPUs = require('node:os').availableParallelism();
const process = require('node:process');

if (cluster.isPrimary) {
  console.log(`Primary ${process.pid} is running`);

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

  cluster.on('exit', (worker, code, signal) => {
    console.log(`worker ${worker.process.pid} died`);
  });
} else {
  // Workers can share any TCP connection
  // In this case it is an HTTP server
  http.createServer((req, res) => {
    res.writeHead(200);
    res.end('hello world\n');
  }).listen(8000);

  console.log(`Worker ${process.pid} started`);
}

现在,运行 Node.js 将在工作进程之间共享 8000 端口。

$ node server.js
Primary 3596 is running
Worker 4324 started
Worker 4520 started
Worker 6056 started
Worker 5644 started 

在 Windows 上,目前还无法在工作进程中设置命名管道服务器。

工作原理#

工作进程是使用 child_process.fork() 方法派生的,因此它们可以通过 IPC 与父进程通信,并来回传递服务器句柄。

cluster 模块支持两种分发传入连接的方法。

第一种(也是除 Windows 外所有平台上的默认方法)是轮询(round-robin)方法,即主进程监听一个端口,接受新连接,并以轮询方式将它们分发给工作进程,同时内置了一些智能机制以避免工作进程过载。

第二种方法是主进程创建监听套接字并将其发送给感兴趣的工作进程。然后,工作进程直接接受传入的连接。

理论上,第二种方法应该能提供最佳性能。然而在实践中,由于操作系统调度程序的变幻莫测,分发往往非常不均衡。曾观察到,在总共八个进程中,超过 70% 的连接最终都落在了仅仅两个进程上。

因为 server.listen() 将大部分工作交给了主进程,所以在以下三种情况下,普通 Node.js 进程和集群工作进程的行为会有所不同:

  1. server.listen({fd: 7}) 因为消息被传递给主进程,所以监听的是父进程中的文件描述符 7,句柄被传递给工作进程,而不是监听工作进程自己认为的文件描述符 7 所引用的内容。
  2. server.listen(handle) 显式地监听句柄将导致工作进程使用提供的句柄,而不是与主进程通信。
  3. server.listen(0) 通常情况下,这会导致服务器在随机端口上监听。然而,在集群中,每个工作进程每次执行 listen(0) 时都会收到相同的“随机”端口。实质上,端口在第一次是随机的,但此后是可预测的。要监听一个唯一的端口,请根据集群工作进程的 ID 生成一个端口号。

Node.js 不提供路由逻辑。因此,设计应用程序时,对于像会话和登录这样的东西,不要过分依赖内存中的数据对象,这一点很重要。

因为工作进程都是独立的进程,可以根据程序的需要被杀死或重新派生,而不会影响其他工作进程。只要还有一些工作进程存活,服务器就会继续接受连接。如果没有存活的工作进程,现有连接将被断开,新的连接将被拒绝。然而,Node.js 不会自动管理工作进程的数量。应用程序有责任根据自身需求来管理工作进程池。

虽然 node:cluster 模块的一个主要用例是网络编程,但它也可以用于其他需要工作进程的场景。

类: Worker#

一个 Worker 对象包含关于一个工作进程的所有公共信息和方法。在主进程中,可以通过 cluster.workers 获取。在工作进程中,可以通过 cluster.worker 获取。

事件: 'disconnect'#

类似于 cluster.on('disconnect') 事件,但特定于此工作进程。

cluster.fork().on('disconnect', () => {
  // Worker has disconnected
}); 

事件: 'error'#

此事件与 child_process.fork() 提供的事件相同。

在工作进程内部,也可以使用 process.on('error')

事件: 'exit'#

  • code <number> 如果是正常退出,则为退出码。
  • signal <string> 导致进程被杀死的信号名称(例如 'SIGHUP')。

类似于 cluster.on('exit') 事件,但特定于此工作进程。

import cluster from 'node:cluster';

if (cluster.isPrimary) {
  const worker = cluster.fork();
  worker.on('exit', (code, signal) => {
    if (signal) {
      console.log(`worker was killed by signal: ${signal}`);
    } else if (code !== 0) {
      console.log(`worker exited with error code: ${code}`);
    } else {
      console.log('worker success!');
    }
  });
}const cluster = require('node:cluster');

if (cluster.isPrimary) {
  const worker = cluster.fork();
  worker.on('exit', (code, signal) => {
    if (signal) {
      console.log(`worker was killed by signal: ${signal}`);
    } else if (code !== 0) {
      console.log(`worker exited with error code: ${code}`);
    } else {
      console.log('worker success!');
    }
  });
}

事件: 'listening'#

类似于 cluster.on('listening') 事件,但特定于此工作进程。

cluster.fork().on('listening', (address) => {
  // Worker is listening
});cluster.fork().on('listening', (address) => {
  // Worker is listening
});

它不会在工作进程中触发。

事件: 'message'#

类似于 cluster'message' 事件,但特定于此工作进程。

在工作进程内部,也可以使用 process.on('message')

参见 process 事件: 'message'

下面是一个使用消息系统的示例。它在主进程中记录工作进程接收到的 HTTP 请求数量:

import cluster from 'node:cluster';
import http from 'node:http';
import { availableParallelism } from 'node:os';
import process from 'node:process';

if (cluster.isPrimary) {

  // Keep track of http requests
  let numReqs = 0;
  setInterval(() => {
    console.log(`numReqs = ${numReqs}`);
  }, 1000);

  // Count requests
  function messageHandler(msg) {
    if (msg.cmd && msg.cmd === 'notifyRequest') {
      numReqs += 1;
    }
  }

  // Start workers and listen for messages containing notifyRequest
  const numCPUs = availableParallelism();
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  for (const id in cluster.workers) {
    cluster.workers[id].on('message', messageHandler);
  }

} else {

  // Worker processes have a http server.
  http.Server((req, res) => {
    res.writeHead(200);
    res.end('hello world\n');

    // Notify primary about the request
    process.send({ cmd: 'notifyRequest' });
  }).listen(8000);
}const cluster = require('node:cluster');
const http = require('node:http');
const numCPUs = require('node:os').availableParallelism();
const process = require('node:process');

if (cluster.isPrimary) {

  // Keep track of http requests
  let numReqs = 0;
  setInterval(() => {
    console.log(`numReqs = ${numReqs}`);
  }, 1000);

  // Count requests
  function messageHandler(msg) {
    if (msg.cmd && msg.cmd === 'notifyRequest') {
      numReqs += 1;
    }
  }

  // Start workers and listen for messages containing notifyRequest
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  for (const id in cluster.workers) {
    cluster.workers[id].on('message', messageHandler);
  }

} else {

  // Worker processes have a http server.
  http.Server((req, res) => {
    res.writeHead(200);
    res.end('hello world\n');

    // Notify primary about the request
    process.send({ cmd: 'notifyRequest' });
  }).listen(8000);
}

事件: 'online'#

类似于 cluster.on('online') 事件,但特定于此工作进程。

cluster.fork().on('online', () => {
  // Worker is online
}); 

它不会在工作进程中触发。

worker.disconnect()#

在工作进程中,此函数将关闭所有服务器,等待这些服务器上的 'close' 事件,然后断开 IPC 通道。

在主进程中,会向工作进程发送一个内部消息,使其在自身上调用 .disconnect()

会导致 .exitedAfterDisconnect 被设置。

服务器关闭后,它将不再接受新的连接,但连接可能会被任何其他正在监听的工作进程接受。现有连接将被允许正常关闭。当不再有连接存在时,参见 server.close(),与工作进程的 IPC 通道将关闭,使其能够优雅地退出。

上述内容*仅*适用于服务器连接,客户端连接不会被工作进程自动关闭,并且 disconnect 不会等待它们关闭后再退出。

在工作进程中,存在 process.disconnect,但它不是此函数;它是 disconnect()

因为长时间存活的服务器连接可能会阻止工作进程断开连接,所以发送一个消息可能会很有用,以便应用程序可以采取特定操作来关闭它们。实现一个超时机制也可能很有用,如果在一段时间后 'disconnect' 事件仍未触发,则杀死该工作进程。

if (cluster.isPrimary) {
  const worker = cluster.fork();
  let timeout;

  worker.on('listening', (address) => {
    worker.send('shutdown');
    worker.disconnect();
    timeout = setTimeout(() => {
      worker.kill();
    }, 2000);
  });

  worker.on('disconnect', () => {
    clearTimeout(timeout);
  });

} else if (cluster.isWorker) {
  const net = require('node:net');
  const server = net.createServer((socket) => {
    // Connections never end
  });

  server.listen(8000);

  process.on('message', (msg) => {
    if (msg === 'shutdown') {
      // Initiate graceful close of any connections to server
    }
  });
} 

worker.exitedAfterDisconnect#

如果工作进程因 .disconnect() 而退出,则此属性为 true。如果工作进程以任何其他方式退出,则为 false。如果工作进程尚未退出,则为 undefined

布尔值 worker.exitedAfterDisconnect 允许区分自愿退出和意外退出,主进程可以根据此值选择不重新派生工作进程。

cluster.on('exit', (worker, code, signal) => {
  if (worker.exitedAfterDisconnect === true) {
    console.log('Oh, it was just voluntary – no need to worry');
  }
});

// kill worker
worker.kill(); 

worker.id#

每个新的工作进程都会被赋予一个唯一的 ID,此 ID 存储在 id 中。

当一个工作进程存活时,这个 ID 是它在 cluster.workers 中的索引键。

worker.isConnected()#

如果工作进程通过其 IPC 通道连接到其主进程,此函数返回 true,否则返回 false。一个工作进程在被创建后即连接到其主进程。在 'disconnect' 事件被触发后,它会断开连接。

worker.isDead()#

如果工作进程的进程已经终止(无论是由于退出还是被信号杀死),此函数返回 true。否则,返回 false

import cluster from 'node:cluster';
import http from 'node:http';
import { availableParallelism } from 'node:os';
import process from 'node:process';

const numCPUs = availableParallelism();

if (cluster.isPrimary) {
  console.log(`Primary ${process.pid} is running`);

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

  cluster.on('fork', (worker) => {
    console.log('worker is dead:', worker.isDead());
  });

  cluster.on('exit', (worker, code, signal) => {
    console.log('worker is dead:', worker.isDead());
  });
} else {
  // Workers can share any TCP connection. In this case, it is an HTTP server.
  http.createServer((req, res) => {
    res.writeHead(200);
    res.end(`Current process\n ${process.pid}`);
    process.kill(process.pid);
  }).listen(8000);
}const cluster = require('node:cluster');
const http = require('node:http');
const numCPUs = require('node:os').availableParallelism();
const process = require('node:process');

if (cluster.isPrimary) {
  console.log(`Primary ${process.pid} is running`);

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

  cluster.on('fork', (worker) => {
    console.log('worker is dead:', worker.isDead());
  });

  cluster.on('exit', (worker, code, signal) => {
    console.log('worker is dead:', worker.isDead());
  });
} else {
  // Workers can share any TCP connection. In this case, it is an HTTP server.
  http.createServer((req, res) => {
    res.writeHead(200);
    res.end(`Current process\n ${process.pid}`);
    process.kill(process.pid);
  }).listen(8000);
}

worker.kill([signal])#

  • signal <string> 发送给工作进程的 kill 信号的名称。默认值: 'SIGTERM'

此函数将杀死工作进程。在主进程中,它通过断开 worker.process 的连接来实现,并在断开连接后,用 signal 杀死它。在工作进程中,它通过用 signal 杀死进程来实现。

kill() 函数会杀死工作进程而不等待其优雅断开,其行为与 worker.process.kill() 相同。

为了向后兼容,此方法也别名为 worker.destroy()

在工作进程中,存在 process.kill(),但它不是此函数;它是 kill()

worker.process#

所有工作进程都是使用 child_process.fork() 创建的,此函数返回的对象存储为 .process。在工作进程中,存储的是全局的 process

参见:子进程模块

如果 process 上发生 'disconnect' 事件且 .exitedAfterDisconnect 不为 true,工作进程将调用 process.exit(0)。这可以防止意外断开连接。

worker.send(message[, sendHandle[, options]][, callback])#

  • message <Object>
  • sendHandle <Handle>
  • options <Object> 如果存在 options 参数,它是一个用于参数化发送某些类型句柄的对象。options 支持以下属性:
    • keepOpen <boolean> 一个可以在传递 net.Socket 实例时使用的值。当为 true 时,套接字在发送进程中保持打开状态。默认值: false
  • callback <Function>
  • 返回:<boolean>

向工作进程或主进程发送消息,可选择性地附带一个句柄。

在主进程中,这会向一个特定的工作进程发送消息。它与 ChildProcess.send() 完全相同。

在工作进程中,这会向主进程发送消息。它与 process.send() 完全相同。

此示例将回显来自主进程的所有消息:

if (cluster.isPrimary) {
  const worker = cluster.fork();
  worker.send('hi there');

} else if (cluster.isWorker) {
  process.on('message', (msg) => {
    process.send(msg);
  });
} 

事件: 'disconnect'#

在工作进程 IPC 通道断开后触发。这可能发生在工作进程正常退出、被杀死或被手动断开连接(例如使用 worker.disconnect())时。

'disconnect''exit' 事件之间可能会有延迟。这些事件可用于检测进程是否在清理过程中卡住,或者是否存在长时间存活的连接。

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

事件: 'exit'#

当任何一个工作进程死亡时,cluster 模块将触发 'exit' 事件。

这可以用于通过再次调用 .fork() 来重启工作进程。

cluster.on('exit', (worker, code, signal) => {
  console.log('worker %d died (%s). restarting...',
              worker.process.pid, signal || code);
  cluster.fork();
}); 

参见 child_process 事件: 'exit'

事件: 'fork'#

当一个新的工作进程被派生时,cluster 模块将触发一个 'fork' 事件。这可以用于记录工作进程活动,并创建自定义超时。

const timeouts = [];
function errorMsg() {
  console.error('Something must be wrong with the connection ...');
}

cluster.on('fork', (worker) => {
  timeouts[worker.id] = setTimeout(errorMsg, 2000);
});
cluster.on('listening', (worker, address) => {
  clearTimeout(timeouts[worker.id]);
});
cluster.on('exit', (worker, code, signal) => {
  clearTimeout(timeouts[worker.id]);
  errorMsg();
}); 

事件: 'listening'#

在工作进程中调用 listen() 之后,当服务器上触发 'listening' 事件时,在主进程的 cluster 上也会触发一个 'listening' 事件。

事件处理程序执行时带有两个参数,worker 包含工作进程对象,address 对象包含以下连接属性:addressportaddressType。如果工作进程正在监听多个地址,这将非常有用。

cluster.on('listening', (worker, address) => {
  console.log(
    `A worker is now connected to ${address.address}:${address.port}`);
}); 

addressType 是以下之一:

  • 4 (TCPv4)
  • 6 (TCPv6)
  • -1 (Unix 域套接字)
  • 'udp4''udp6' (UDPv4 或 UDPv6)

事件: 'message'#

当集群主进程收到来自任何工作进程的消息时触发。

参见 child_process 事件: 'message'

事件: 'online'#

派生一个新的工作进程后,该工作进程应该响应一个在线消息。当主进程收到在线消息时,它将触发此事件。'fork''online' 的区别在于,fork 是在主进程派生工作进程时触发的,而 'online' 是在工作进程运行时触发的。

cluster.on('online', (worker) => {
  console.log('Yay, the worker responded after it was forked');
}); 

事件: 'setup'#

每次调用 .setupPrimary() 时都会触发。

settings 对象是调用 .setupPrimary() 时的 cluster.settings 对象,并且仅供参考,因为在单个 tick 中可以多次调用 .setupPrimary()

如果准确性很重要,请使用 cluster.settings

cluster.disconnect([callback])#

  • callback <Function> 当所有工作进程都断开连接并且句柄都关闭时调用。

cluster.workers 中的每个工作进程调用 .disconnect()

当它们断开连接时,所有内部句柄将被关闭,如果没有其他事件在等待,则允许主进程正常退出。

该方法接受一个可选的回调参数,该参数将在完成后被调用。

这只能从主进程调用。

cluster.fork([env])#

派生一个新的工作进程。

这只能从主进程调用。

cluster.isMaster#

稳定性: 0 - 废弃

cluster.isPrimary 的废弃别名。

cluster.isPrimary#

如果进程是主进程,则为 True。这是由 process.env.NODE_UNIQUE_ID 决定的。如果 process.env.NODE_UNIQUE_ID 是 undefined,则 isPrimarytrue

cluster.isWorker#

如果进程不是主进程,则为 True(它是 cluster.isPrimary 的否定)。

cluster.schedulingPolicy#

调度策略,可以是用于轮询的 cluster.SCHED_RR,或者交由操作系统处理的 cluster.SCHED_NONE。这是一个全局设置,一旦第一个工作进程被派生,或者 .setupPrimary() 被调用(以先发生者为准),该设置实际上就被冻结了。

除了 Windows,SCHED_RR 是所有操作系统上的默认值。一旦 libuv 能够有效分发 IOCP 句柄而不会导致大的性能损失,Windows 将更改为 SCHED_RR

cluster.schedulingPolicy 也可以通过 NODE_CLUSTER_SCHED_POLICY 环境变量来设置。有效值为 'rr''none'

cluster.settings#

  • 类型:<Object>
    • execArgv <string[]> 传递给 Node.js 可执行文件的字符串参数列表。默认值: process.execArgv
    • exec <string> 工作进程文件的路径。默认值: process.argv[1]
    • args <string[]> 传递给工作进程的字符串参数。默认值: process.argv.slice(2)
    • cwd <string> 工作进程的当前工作目录。默认值: undefined(从父进程继承)。
    • serialization <string> 指定用于在进程之间发送消息的序列化类型。可能的值是 'json''advanced'。更多详情请参见 child_process 的高级序列化默认值: false
    • silent <boolean> 是否将输出发送到父进程的 stdio。默认值: false
    • stdio <Array> 配置派生进程的 stdio。因为 cluster 模块依赖 IPC 来运行,这个配置必须包含一个 'ipc' 条目。当提供此选项时,它会覆盖 silent。参见 child_process.spawn()stdio
    • uid <number> 设置进程的用户标识。(参见 setuid(2).)
    • gid <number> 设置进程的组标识。(参见 setgid(2).)
    • inspectPort <number> | <Function> 设置工作进程的检查器端口。这可以是一个数字,或者一个不带参数并返回数字的函数。默认情况下,每个工作进程都会获得自己的端口,从主进程的 process.debugPort 开始递增。
    • windowsHide <boolean> 隐藏在 Windows 系统上通常会创建的派生进程的控制台窗口。默认值: false

调用 .setupPrimary()(或 .fork())之后,此设置对象将包含设置,包括默认值。

不应手动更改或设置此对象。

cluster.setupMaster([settings])#

稳定性: 0 - 废弃

.setupPrimary() 的废弃别名。

cluster.setupPrimary([settings])#

setupPrimary 用于更改默认的 'fork' 行为。一旦调用,这些设置将存在于 cluster.settings 中。

任何设置更改仅影响未来的 .fork() 调用,对已经运行的工作进程没有影响。

工作进程中唯一不能通过 .setupPrimary() 设置的属性是传递给 .fork()env

上述默认值仅适用于首次调用;后续调用的默认值是调用 cluster.setupPrimary() 时的当前值。

import cluster from 'node:cluster';

cluster.setupPrimary({
  exec: 'worker.js',
  args: ['--use', 'https'],
  silent: true,
});
cluster.fork(); // https worker
cluster.setupPrimary({
  exec: 'worker.js',
  args: ['--use', 'http'],
});
cluster.fork(); // http workerconst cluster = require('node:cluster');

cluster.setupPrimary({
  exec: 'worker.js',
  args: ['--use', 'https'],
  silent: true,
});
cluster.fork(); // https worker
cluster.setupPrimary({
  exec: 'worker.js',
  args: ['--use', 'http'],
});
cluster.fork(); // http worker

这只能从主进程调用。

cluster.worker#

对当前工作进程对象的引用。在主进程中不可用。

import cluster from 'node:cluster';

if (cluster.isPrimary) {
  console.log('I am primary');
  cluster.fork();
  cluster.fork();
} else if (cluster.isWorker) {
  console.log(`I am worker #${cluster.worker.id}`);
}const cluster = require('node:cluster');

if (cluster.isPrimary) {
  console.log('I am primary');
  cluster.fork();
  cluster.fork();
} else if (cluster.isWorker) {
  console.log(`I am worker #${cluster.worker.id}`);
}

cluster.workers#

一个存储活动工作进程对象的哈希表,以 id 字段为键。这使得遍历所有工作进程变得容易。它仅在主进程中可用。

一个工作进程在断开连接*并且*退出后会从 cluster.workers 中移除。这两个事件的顺序无法预先确定。但是,可以保证从 cluster.workers 列表中移除的操作发生在最后一个 'disconnect''exit' 事件触发之前。

import cluster from 'node:cluster';

for (const worker of Object.values(cluster.workers)) {
  worker.send('big announcement to all workers');
}const cluster = require('node:cluster');

for (const worker of Object.values(cluster.workers)) {
  worker.send('big announcement to all workers');
}