Node.js v12.2.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 已启动

工作原理#

查看v10.x中文文档

The worker processes are spawned using the child_process.fork() method, so that they can communicate with the parent via IPC and pass server handles back and forth.

The cluster module supports two methods of distributing incoming connections.

The first one (and the default one on all platforms except Windows), is the round-robin approach, where the master process listens on a port, accepts new connections and distributes them across the workers in a round-robin fashion, with some built-in smarts to avoid overloading a worker process.

The second approach is where the master process creates the listen socket and sends it to interested workers. The workers then accept incoming connections directly.

The second approach should, in theory, give the best performance. In practice however, distribution tends to be very unbalanced due to operating system scheduler vagaries. Loads have been observed where over 70% of all connections ended up in just two processes, out of a total of eight.

Because server.listen() hands off most of the work to the master process, there are three cases where the behavior between a normal Node.js process and a cluster worker differs:

  1. server.listen({fd: 7}) Because the message is passed to the master, file descriptor 7 in the parent will be listened on, and the handle passed to the worker, rather than listening to the worker's idea of what the number 7 file descriptor references.
  2. server.listen(handle) Listening on handles explicitly will cause the worker to use the supplied handle, rather than talk to the master process.
  3. server.listen(0) Normally, this will cause servers to listen on a random port. However, in a cluster, each worker will receive the same "random" port each time they do listen(0). In essence, the port is random the first time, but predictable thereafter. To listen on a unique port, generate a port number based on the cluster worker ID.

Node.js does not provide routing logic. It is, therefore important to design an application such that it does not rely too heavily on in-memory data objects for things like sessions and login.

Because workers are all separate processes, they can be killed or re-spawned depending on a program's needs, without affecting other workers. As long as there are some workers still alive, the server will continue to accept connections. If no workers are alive, existing connections will be dropped and new connections will be refused. Node.js does not automatically manage the number of workers, however. It is the application's responsibility to manage the worker pool based on its own needs.

Although a primary use case for the cluster module is networking, it can also be used for other use cases requiring worker processes.

Worker 类#

中英对照提交修改

Worker对象包含了关于工作进程的所有public信息和方法。

在一个主进程里,可以使用cluster.workers来获取Worker对象。

在一个工作进程里,可以使用cluster.worker来获取Worker对象。

'disconnect' 事件#

中英对照提交修改

虽然与 cluster.on('disconnect')事件 是相似的,但是这个进程又有其他特征。

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

'error' 事件#

中英对照提交修改

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

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

'exit' 事件#

中英对照提交修改

  • code <number> 若正常退出,表示退出代码.
  • signal <string> 引发进程被kill的信号名称(如'SIGHUP').

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

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
});

本事件不会在工作进程内触发。

'message' 事件#

查看v10.x中文文档

Similar to the 'message' event of cluster, but specific to this worker.

Within a worker, process.on('message') may also be used.

See process event: 'message'.

Here is an example using the message system. It keeps a count in the master process of the number of HTTP requests received by the workers:

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

if (cluster.isMaster) {

  // 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 = 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 {

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

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

'online' 事件#

中英对照提交修改

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

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

本事件不会在工作进程内部被触发。

worker.disconnect()#

查看v10.x中文文档

In a worker, this function will close all servers, wait for the 'close' event on those servers, and then disconnect the IPC channel.

In the master, an internal message is sent to the worker causing it to call .disconnect() on itself.

Causes .exitedAfterDisconnect to be set.

Note that after a server is closed, it will no longer accept new connections, but connections may be accepted by any other listening worker. Existing connections will be allowed to close as usual. When no more connections exist, see server.close(), the IPC channel to the worker will close allowing it to die gracefully.

The above applies only to server connections, client connections are not automatically closed by workers, and disconnect does not wait for them to close before exiting.

Note that in a worker, process.disconnect exists, but it is not this function, it is disconnect.

Because long living server connections may block workers from disconnecting, it may be useful to send a message, so application specific actions may be taken to close them. It also may be useful to implement a timeout, killing a worker if the 'disconnect' event has not been emitted after some time.

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) => {
    // Connections never end
  });

  server.listen(8000);

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

worker.exitedAfterDisconnect#

中英对照提交修改

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

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

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

// 关闭 worker
worker.kill();

worker.id#

中英对照提交修改

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

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

worker.isConnected()#

中英对照提交修改

当工作进程通过IPC管道连接至主进程时,这个方法返回true,否则返回false

一个工作进程在创建后会自动连接到它的主进程,当'disconnect' 事件被触发时才会断开连接。

worker.isDead()#

中英对照提交修改

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

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

中英对照提交修改

  • signal <string> 被发送kill信号的工作进程名称。

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

将导致.exitedAfterDisconnect被设置。

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

需要注意的是,在工作进程中有一个方法process.kill() ,这个方法与本方法不同,本方法是kill

worker.process#

中英对照提交修改

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

详见:Child Process module

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

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

中英对照提交修改

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

主进程调用这个方法会发送消息给具体的工作进程。还有一个等价的方法是ChildProcess.send()

工作进程调用这个方法会发送消息给主进程。还有一个等价方法是process.send()

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

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

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

'disconnect' 事件#

中英对照提交修改

在工作进程的IPC管道被断开后触发本事件。可能导致事件触发的原因包括:工作进程优雅地退出、被kill或手动断开连接(如调用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 event: 'exit'][]。

'fork' 事件#

中英对照提交修改

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

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()后,工作进程上的server会触发'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 domain socket)
  • "udp4" or "udp6" (UDP v4 or v6)

'message' 事件#

中英对照提交修改

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

详见: [child_process event: 'message'][]。

和文档情况相反的是:在Node.js v6.0版本之前,这个事件仅仅接受两个参数:消息和handle,而没有工作进程对象。

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

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

'online' 事件#

中英对照提交修改

当新建一个工作进程后,工作进程应当响应一个online消息给主进程。当主进程收到online消息后触发这个事件。 'fork' 事件和 'online'事件的不同之处在于,前者是在主进程新建工作进程后触发,而后者是在工作进程运行的时候触发。

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

'setup' 事件#

中英对照提交修改

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

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

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

cluster.disconnect([callback])#

中英对照提交修改

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

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

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

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

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

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 handle,而不会导致严重的性能冲击的话,Windows系统也会更改为SCHED_RR

cluster.schedulingPolicy 可以通过设置NODE_CLUSTER_SCHED_POLICY环境变量来实现。这个环境变量的有效值包括"rr""none"

cluster.settings#

中英对照提交修改

  • <Object>   execArgv <Array> 传递给Node.js可执行文件的参数列表。 (Default=process.execArgv)   exec <string> worker文件路径。 (Default=process.argv[1])   args <Array> 传递给worker的参数。(Default=process.argv.slice(2))   silent <boolean> 是否需要发送输出值父进程的stdio。(Default=false)   stdio <Array> 配置fork进程的stdio。  由于cluster模块运行依赖于IPC,这个配置必须包含'ipc'。当提供了这个选项后,将撤销silent。   uid <number> 设置进程的user标识符。 (见 setuid(2).)  * gid <number> 设置进程的group标识符。 (见 setgid(2).)

    • inspectPort <number> | <Function> Sets inspector port of worker. This can be a number, or a function that takes no arguments and returns a number. By default each worker gets its own port, incremented from the master's process.debugPort.

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

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

cluster.setupMaster([settings])#

中英对照提交修改

用于修改默认'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 worker
cluster.setupMaster({
  exec: 'worker.js',
  args: ['--use', 'http']
});
cluster.fork(); // http worker

只能由主进程调用。

cluster.worker#

中英对照提交修改

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

const cluster = require('cluster');

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

cluster.workers#

中英对照提交修改

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

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

// Go through all workers
function eachWorker(callback) {
  for (const id in cluster.workers) {
    callback(cluster.workers[id]);
  }
}
eachWorker((worker) => {
  worker.send('big announcement to all workers');
});

使用工作进程的id来进行定位索引是最方便的!

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