Node.js v12.9.0 文档


cluster(集群)#

中英对照提交修改

稳定性: 2 - 稳定

单个 Node.js 实例运行在单个线程中。 为了充分利用多核系统,有时需要启用一组 Node.js 进程去处理负载任务。

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

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} 已退出`);
  });
} else {
  // 工作进程可以共享任何 TCP 连接。
  // 在本例子中,共享的是 HTTP 服务器。
  http.createServer((req, res) => {
    res.writeHead(200);
    res.end('你好世界\n');
  }).listen(8000);

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

运行代码,则工作进程会共享 8000 端口:

$ node server.js
主进程 3596 正在运行
工作进程 4324 已启动
工作进程 4520 已启动
工作进程 6056 已启动
工作进程 5644 已启动

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

工作原理#

中英对照提交修改

工作进程由 child_process.fork() 方法创建,因此它们可以使用 IPC 和父进程通信,从而使各进程交替处理连接服务。

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

第一种方法(也是除 Windows 外所有平台的默认方法)是循环法,由主进程负责监听端口,接收新连接后再将连接循环分发给工作进程,在分发中使用了一些内置技巧防止工作进程任务过载。

第二种方法是,主进程创建监听 socket 后发送给感兴趣的工作进程,由工作进程负责直接接收连接。

理论上第二种方法应该是效率最佳的。 但在实际情况下,由于操作系统调度机制的难以捉摸,会使分发变得不稳定。 可能会出现八个进程中有两个分担了 70% 的负载。

因为 server.listen() 将大部分工作交给主进程完成,因此导致普通 Node.js 进程与 cluster 工作进程差异的情况有三种:

  1. server.listen({fd: 7}) 因为消息会被传给主进程,所以父进程中的文件描述符 7 将会被监听并将句柄传给工作进程,而不是监听文件描述符 7 指向的工作进程。
  2. server.listen(handle) 显式地监听句柄,会导致工作进程直接使用该句柄,而不是和主进程通信。
  3. server.listen(0) 正常情况下,这种调用会导致 server 在随机端口上监听。 但在 cluster 模式中,所有工作进程每次调用 listen(0) 时会收到相同的“随机”端口。 实质上,这种端口只在第一次分配时随机,之后就变得可预料。 如果要使用独立端口的话,应该根据工作进程的 ID 来生成端口号。

Node.js 不支持路由逻辑。 因此在设计应用时,不应该过分依赖内存数据对象,例如 session 和登陆等。

由于各工作进程是独立的进程,它们可以根据需要随时关闭或重新生成,而不影响其他进程的正常运行。 只要有存活的工作进程,服务器就可以继续处理连接。 如果没有存活的工作进程,现有连接会丢失,新的连接也会被拒绝。 Node.js 不会自动管理工作进程的数量,而应该由具体的应用根据实际需要来管理进程池。

虽然 cluster 模块主要用于网络相关的情况,但同样可以用于其他需要工作进程的情况。

Worker 类#

中英对照提交修改

Worker 对象包含了关于工作进程的所有公共信息和方法。 在主进程里,可以使用 cluster.workers 来获取它。 在工作进程里,可以使用 cluster.worker 来获取它。

'disconnect' 事件#

中英对照提交修改

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

cluster.fork().on('disconnect', () => {
  // 工作进程已断开连接。
});

'error' 事件#

中英对照提交修改

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

在一个工作进程中,也可以使用 process.on('error')

'exit' 事件#

中英对照提交修改

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

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

const worker = cluster.fork();
worker.on('exit', (code, signal) => {
  if (signal) {
    console.log(`工作进程已被信号 ${signal} 杀死`);
  } else if (code !== 0) {
    console.log(`工作进程退出,退出码: ${code}`);
  } else {
    console.log('工作进程成功退出');
  }
});

'listening' 事件#

中英对照提交修改

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

cluster.fork().on('listening', (address) => {
  // 工作进程正在监听。
});

此事件不会在工作进程中触发。

'message' 事件#

中英对照提交修改

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

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

参阅 process event: 'message'

以下是一个使用消息系统的示例。 它在主进程中对工作进程接收的 HTTP 请求数量保持计数:

const cluster = require('cluster');
const http = require('http');

if (cluster.isMaster) {

  // 跟踪 http 请求。
  let numReqs = 0;
  setInterval(() => {
    console.log(`请求的数量 = ${numReqs}`);
  }, 1000);

  // 对请求计数。
  function messageHandler(msg) {
    if (msg.cmd && msg.cmd === 'notifyRequest') {
      numReqs += 1;
    }
  }

  // 启动 worker 并监听包含 notifyRequest 的消息。
  const numCPUs = require('os').cpus().length;
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

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

} else {

  // 工作进程有一个 http 服务器。
  http.Server((req, res) => {
    res.writeHead(200);
    res.end('你好世界\n');

    // 通知主进程接收到了请求。
    process.send({ cmd: 'notifyRequest' });
  }).listen(8000);
}

'online' 事件#

中英对照提交修改

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

cluster.fork().on('online', () => {
  // 工作进程已上线。
});

此事件不会在工作进程中触发。

worker.disconnect()#

中英对照提交修改

在一个工作进程内,调用此方法会关闭所有的 server,并等待这些 server 的 'close' 事件执行,然后关闭 IPC 管道。

在主进程内,会给工作进程发送一个内部消息,导致工作进程自身调用 .disconnect()

会设置 .exitedAfterDisconnect

当一个 server 关闭后,它将不再接收新的连接,但新连接会被其他正在监听的工作进程接收。 已建立的连接可以正常关闭。 当所有连接都关闭后,通往该工作进程的 IPC 管道将会关闭,允许工作进程优雅地死掉,详见 server.close()

以上情况只针对服务端连接,工作进程不会自动关闭客户端连接,disconnect 方法在退出前并不会等待客户端连接关闭。

在工作进程中,也存在 process.disconnect,但它不是这个函数,它是 disconnect()

因为长时间运行的服务端连接可能阻止工作进程断开连接,可以采用发送消息的方法,让应用采取相应的动作来关闭连接。 也可以通过设置超时,当 'disconnect' 事件在某段时间后仍没有触发时关闭工作进程。

if (cluster.isMaster) {
  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('net');
  const server = net.createServer((socket) => {
    // 连接永远不会结束。
  });

  server.listen(8000);

  process.on('message', (msg) => {
    if (msg === 'shutdown') {
      // 将所有与服务器的连接优雅地关闭。
    }
  });
}

worker.exitedAfterDisconnect#

中英对照提交修改

当调用 .kill() 或者 .disconnect() 方法时被设置,在这之前都是 undefined

worker.exitedAfterDisconnect 可以用于区分自发退出还是被动退出,主进程可以根据这个值决定是否重新衍生新的工作进程。

cluster.on('exit', (worker, code, signal) => {
  if (worker.exitedAfterDisconnect === true) {
    console.log('这是自发退出,无需担心。');
  }
});

// 杀死工作进程。
worker.kill();

worker.id#

中英对照提交修改

每一个新衍生的工作进程都会被赋予自己独一无二的编号,这个编号就是储存在 id 里面。

当工作进程还存活时,这个编号可以作为在 cluster.workers 中的索引。

worker.isConnected()#

中英对照提交修改

当工作进程通过 IPC 管道连接至主进程时,这个方法返回 true,否则返回 false。 一个工作进程在创建后会自动连接到它的主进程。 当 'disconnect' 事件被触发时才会断开连接。

worker.isDead()#

中英对照提交修改

当工作进程被终止时(包括自动退出或被发送信号),这个方法返回 true。 否则,返回 false

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('fork', (worker) => {
    console.log('工作进程已关闭:', worker.isDead());
  });

  cluster.on('exit', (worker, code, signal) => {
    console.log('工作进程已关闭:', worker.isDead());
  });
} else {
  // 工作进程可以共享任何 TCP 连接。在这种情况下,它是一个 HTTP 服务器。
  http.createServer((req, res) => {
    res.writeHead(200);
    res.end(`当前进程\n ${process.pid}`);
    process.kill(process.pid);
  }).listen(8000);
}

worker.kill([signal='SIGTERM'])#

中英对照提交修改

  • signal <string> 发送给工作进程的杀死信号的名称。

这个方法将会杀死工作进程。 在主进程中,通过断开与 worker.process 的连接来实现,一旦断开连接后,通过 signal 来杀死工作进程。 在工作进程中,通过断开 IPC 管道来实现,然后以代码 0 退出进程。

因为 kill() 会尝试正常地断开工作进程,所以很容易无限期地等待断开连接完成。 例如,如果工作进程进入无限循环,则永远不会发生正常断开连接。 如果不需要正常的断开连接行为,请使用 worker.process.kill()

将导致 .exitedAfterDisconnect 被设置。

为向后兼容,这个方法与 worker.destroy() 等义。

在工作进程中,也存在 process.kill(),但它不是这个函数,它是 kill()

worker.process#

中英对照提交修改

所有的工作进程都是通过 child_process.fork() 来创建的,这个方法返回的对象被存储为 .process。 在工作进程中, process 属于全局对象。

参阅:子进程模块

process 上发生 'disconnect' 事件,并且 .exitedAfterDisconnect 的值不是 true 时,工作进程会调用 process.exit(0)。 这样就可以防止连接意外断开。

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

中英对照提交修改

发送一个消息给工作进程或主进程,也可以附带发送一个句柄。

在主进程中,这会发送消息给特定的工作进程。 相当于 ChildProcess.send()

在工作进程中,这会发送消息给主进程。 相当于 process.send()

这个例子里面,工作进程将主进程发送的消息响应回去:

if (cluster.isMaster) {
  const worker = cluster.fork();
  worker.send('你好');

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

'disconnect' 事件#

中英对照提交修改

在工作进程的 IPC 管道被断开后触发。 可能导致事件触发的原因包括:工作进程优雅地退出、被杀死、或手动断开连接(如调用 worker.disconnect())。

'disconnect''exit' 事件之间可能存在延迟。 这些事件可以用来检测进程是否在清理过程中被卡住,或是否存在长时间运行的连接。

cluster.on('disconnect', (worker) => {
  console.log(`工作进程 #${worker.id} 已断开连接`);
});

'exit' 事件#

中英对照提交修改

当任何一个工作进程关闭的时候,cluster 模块都将会触发 'exit' 事件。

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

cluster.on('exit', (worker, code, signal) => {
  console.log('工作进程 %d 关闭 (%s). 重启中...',
              worker.process.pid, signal || code);
  cluster.fork();
});

参阅 child_process event: 'exit'

'fork' 事件#

中英对照提交修改

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

const timeouts = [];
function errorMsg() {
  console.error('连接出错');
}

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() 后,工作进程上的 server 会触发 'listening' 事件,同时主进程上的 cluster 也会触发 'listening' 事件。

事件句柄使用两个参数来执行,其中 worker 包含了工作进程对象, address 包含了以下的连接属性:addressportaddressType。 当工作进程同时监听多个地址时,这些参数非常有用。

cluster.on('listening', (worker, address) => {
  console.log(
    `工作进程已连接到 ${address.address}:${address.port}`);
});

addressType 可选值包括:

  • 4 (TCPv4)
  • 6 (TCPv6)
  • -1 (Unix 域 socket)
  • 'udp4' or 'udp6' (UDP v4 或 v6)

'message' 事件#

中英对照提交修改

当 cluster 主进程接收任意工作进程发送的消息时触发。

参阅 child_process event: 'message'

在 Node.js v6.0 版本之前,此事件只接受 messagehandle 两个参数,而没有 worker 对象。

如果要兼容旧版本并且不需要工作进程对象的情况下,可以通过判断参数数量来实现兼容。

cluster.on('message', (worker, message, handle) => {
  if (arguments.length === 2) {
    handle = message;
    message = worker;
    worker = undefined;
  }
  // ...
});

'online' 事件#

中英对照提交修改

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

cluster.on('online', (worker) => {
  console.log('工作进程被衍生后响应');
});

'setup' 事件#

中英对照提交修改

每当 .setupMaster() 被调用时触发。

settings 对象是 setupMaster() 被调用时的 cluster.settings 对象,并且只能查询,因为在一个时间点内 .setupMaster() 可以被调用多次。

如果精确度十分重要,则使用 cluster.settings

cluster.disconnect([callback])#

中英对照提交修改

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

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

当所有工作进程断开连接后,所有内部句柄将会关闭,这个时候如果没有等待事件的话,运行主进程优雅地关闭。

这个方法可以选择添加一个回调参数,当结束时会调用这个回调函数。

这个方法只能由主进程调用。

cluster.fork([env])#

中英对照提交修改

衍生出一个新的工作进程。

这只能通过主进程调用。

cluster.isMaster#

中英对照提交修改

如果该进程是主进程,则为 true。 这是由 process.env.NODE_UNIQUE_ID 决定的。 如果 process.env.NODE_UNIQUE_ID 未定义,则 isMastertrue

cluster.isWorker#

中英对照提交修改

如果该进程不是主进程,则为 true(与 cluster.isMaster 相反)。

cluster.schedulingPolicy#

中英对照提交修改

调度策略,包括循环计数的 cluster.SCHED_RR,以及由操作系统决定的 cluster.SCHED_NONE。 这是一个全局设置,当第一个工作进程被衍生或者调用 cluster.setupMaster() 时,都将第一时间生效。

除 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(从父进程继承)。
    • silent <boolean> 是否需要发送输出到父进程的 stdio。默认值: false
    • stdio <Array> 配置衍生的进程的 stdio。 由于 cluster 模块运行依赖于 IPC,这个配置必须包含 'ipc'。如果提供了这个选项,则覆盖 silent
    • uid <number> 设置进程的用户标识符。参阅 setuid(2)
    • gid <number> 设置进程的群组标识符。参阅 setgid(2)
    • inspectPort <number> | <Function> 设置工作进程的检查端口。这可以是一个数字、或不带参数并返回数字的函数。默认情况下,每个工作进程都有自己的端口,从主进程的 process.debugPort 开始递增。
    • windowsHide <boolean> 隐藏衍生的进程的控制台窗口(通常在 Windows 系统上会创建)。默认值: false

调用 .setupMaster()(或 .fork())之后,这个配置对象将会包含这些配置项,包括默认值。

这个对象不打算被修改或手动设置。

cluster.setupMaster([settings])#

中英对照提交修改

setupMaster 用于修改默认的 'fork' 行为。 一旦调用,将会按照 cluster.settings 进行设置。

所有的设置只对后来的 .fork() 调用有效,对之前的工作进程无影响。

唯一无法通过 .setupMaster() 设置的属性是传给 .fork()env 属性。

上述的默认值只在第一次调用时有效,当后续调用时,将采用 cluster.setupMaster() 调用时的当前值。

const cluster = require('cluster');
cluster.setupMaster({
  exec: 'worker.js',
  args: ['--use', 'https'],
  silent: true
});
cluster.fork(); // https 工作进程
cluster.setupMaster({
  exec: 'worker.js',
  args: ['--use', 'http']
});
cluster.fork(); // http 工作进程

这只能由主进程调用。

cluster.worker#

中英对照提交修改

当前工作进程对象的引用。 对于主进程则无效。

const cluster = require('cluster');

if (cluster.isMaster) {
  console.log('这是主进程');
  cluster.fork();
  cluster.fork();
} else if (cluster.isWorker) {
  console.log(`这是工作进程 #${cluster.worker.id}`);
}

cluster.workers#

中英对照提交修改

这是一个哈希表,储存了活跃的工作进程对象,使用 id 作为键名。 这使得可以方便地遍历所有工作进程。 只能在主进程中调用。

工作进程断开连接以及退出后,将会从 cluster.workers 里面移除。 这两个事件的先后顺序并不能预先确定。 但可以保证的是, cluster.workers 的移除工作在 'disconnect''exit' 两个事件中的最后一个触发之前完成。

// 遍历所有工作进程。
function eachWorker(callback) {
  for (const id in cluster.workers) {
    callback(cluster.workers[id]);
  }
}
eachWorker((worker) => {
  worker.send('通知所有工作进程');
});

使用工作进程的唯一 id 是定位工作进程最简单的方式。

socket.on('data', (id) => {
  const worker = cluster.workers[id];
});