在Node.js中的cluster怎么使用

发布时间：2023-07-26 10:51:04 所属栏目：教程来源：未知

导读： 　　为大家详细介绍“Node.js中的cluster怎么使用”，内容详细，步骤清晰，细节处理妥当，希望这篇“Node.js中的cluster怎么使用”文章能帮助大家解决疑惑，下面跟着小

　　为大家详细介绍“Node.js中的cluster怎么使用”，内容详细，步骤清晰，细节处理妥当，希望这篇“Node.js中的cluster怎么使用”文章能帮助大家解决疑惑，下面跟着小编的思路慢慢深入，一起来学习新知识吧。

　　当初使用 cluster 时，一直好奇它是怎么做到多个子进程监听同一个端口而不冲突的，比如下面这段代码：

　　const cluster = require('cluster')

　　const net = require('net')

　　const cpus = require('os').cpus()

　　if (cluster.isPrimary) {

　　 for (let i = 0; i < cpus.length; i++) {

　　    cluster.fork()

　　 }

　　} else {

　　 net

　　    .createServer(function (socket) {

　　      socket.on('data', function (data) {

　　        socket.write(`Reply from ${process.pid}: ` + data.toString())

　　      })

　　      socket.on('end', function () {

　　        console.log('Close')

　　      })

　　      socket.write('Hello!\n')

　　    })

　　    .listen(9999)

　　}

　　该段代码通过父进程 fork 出了多个子进程，且这些子进程都监听了 9999 这个端口并能正常提供服务，这是如何做到的呢？我们来研究一下。

　　准备调试环境

　　学习 Node.js 官方提供库最好的方式当然是调试一下，所以，我们先来准备一下环境。注：本文的操作系统为 macOS Big Sur 11.6.6，其他系统请自行准备相应环境。

　　编译 Node.js

　　下载 Node.js 源码

　　git clone https://github.com/nodejs/node.git

　　然后在下面这两个地方加入断点，方便后面调试用：

　　// lib/internal/cluster/primary.js

　　function queryServer(worker, message) {

　　 debugger;

　　 // Stop processing if worker already disconnecting

　　 if (worker.exitedAfterDisconnect) return;

　　 ...

　　}

　　// lib/internal/cluster/child.js

　　send(message, (reply, handle) => {

　　 debugger

　　 if (typeof obj._setServerData === 'function') obj._setServerData(reply.data)

　　 if (handle) {

　　    // Shared listen socket

　　    shared(reply, {handle, indexesKey, index}, cb)

　　 } else {

　　    // Round-robin.

　　    rr(reply, {indexesKey, index}, cb)

　　 }

　　})

　　进入目录，执行

　　./configure --debug

　　make -j4

　　之后会生成 out/Debug/node

　　准备 IDE 环境

　　使用 vscode 调试，配置好 launch.json 就可以了（其他 IDE 类似，请自行解决）：

　　{

　　 "version": "0.2.0",

　　 "configurations": [

　　    {

　　      "name": "Debug C++",

　　      "type": "cppdbg",

　　      "program": "/Users/youxingzhi/ayou/node/out/Debug/node",

　　      "request": "launch",

　　      "args": ["/Users/youxingzhi/ayou/node/index.js"],

　　      "stopAtEntry": false,

　　      "cwd": "${workspaceFolder}",

　　      "environment": [],

　　      "externalConsole": false,

　　      "MIMode": "lldb"

　　    },

　　    {

　　      "name": "Debug Node",

　　      "type": "node",

　　      "runtimeExecutable": "/Users/youxingzhi/ayou/node/out/Debug/node",

　　      "request": "launch",

　　      "args": ["--expose-internals", "--nolazy"],

　　      "skipFiles": [],

　　      "program": "${workspaceFolder}/index.js"

　　    }

　　 ]

　　}

　　其中第一个是用于调式 C++ 代码（需要安装 C/C++ 插件），第二个用于调式 JS 代码。接下来就可以开始调试了，我们暂时用调式 JS 代码的那个配置就好了。

　　Cluster 源码调试

　　准备好调试代码（为了调试而已，这里启动一个子进程就够了）：

　　debugger

　　const cluster = require('cluster')

　　const net = require('net')

　　if (cluster.isPrimary) {

　　 debugger

　　 cluster.fork()

　　} else {

　　 const server = net.createServer(function (socket) {

　　    socket.on('data', function (data) {

　　      socket.write(`Reply from ${process.pid}: ` + data.toString())

　　    })

　　    socket.on('end', function () {

　　      console.log('Close')

　　    })

　　    socket.write('Hello!\n')

　　 })

　　 debugger

　　 server.listen(9999)

　　}

　　很明显，我们的程序可以分父进程和子进程这两部分来进行分析。

　　首先进入的是父进程：

　　执行 require('cluster') 时，会进入 lib/cluster.js 这个文件：

　　const childOrPrimary = 'NODE_UNIQUE_ID' in process.env ? 'child' : 'primary'

　　module.exports = require(`internal/cluster/${childOrPrimary}`)

　　会根据当前 process.env 上是否有 NODE_UNIQUE_ID 来引入不同的模块，此时是没有的，所以会引入 internal/cluster/primary.js 这个模块：

　　...

　　const cluster = new EventEmitter();

　　...

　　module.exports = cluster

　　const handles = new SafeMap()

　　cluster.isWorker = false

　　cluster.isMaster = true // Deprecated alias. Must be same as isPrimary.

　　cluster.isPrimary = true

　　cluster.Worker = Worker

　　cluster.workers = {}

　　cluster.settings = {}

　　cluster.SCHED_NONE = SCHED_NONE // Leave it to the operating system.

　　cluster.SCHED_RR = SCHED_RR // Primary distributes connections.

　　...

　　cluster.schedulingPolicy = schedulingPolicy

　　cluster.setupPrimary = function (options) {

　　...

　　}

　　// Deprecated alias must be same as setupPrimary

　　cluster.setupMaster = cluster.setupPrimary

　　function setupSettingsNT(settings) {

　　...

　　}

　　function createWorkerProcess(id, env) {

　　 ...

　　}

　　function removeWorker(worker) {

　　 ...

　　}

　　function removeHandlesForWorker(worker) {

　　 ...

　　}

　　cluster.fork = function (env) {

　　 ...

　　}

　　该模块主要是在 cluster 对象上挂载了一些属性和方法，并导出，这些后面回过头再看，我们继续往下调试。往下调试会进入 if (cluster.isPrimary) 分支，代码很简单，仅仅是 fork 出了一个新的子进程而已：

　　// lib/internal/cluster/primary.js

　　cluster.fork = function (env) {

　　 cluster.setupPrimary()

　　 const id = ++ids

　　 const workerProcess = createWorkerProcess(id, env)

　　 const worker = new Worker({

　　    id: id,

　　    process: workerProcess,

　　 })

　　 ...

　　 worker.process.on('internalMessage', internal(worker, onmessage))

　　 process.nextTick(emitForkNT, worker)

　　 cluster.workers[worker.id] = worker

　　 return worker

　　}

　　cluster.setupPrimary()：比较简单，初始化一些参数啥的。

　　createWorkerProcess(id, env)：

　　// lib/internal/cluster/primary.js

　　function createWorkerProcess(id, env) {

　　 const workerEnv = {...process.env, ...env, NODE_UNIQUE_ID: `${id}`}

　　 const execArgv = [...cluster.settings.execArgv]

　　 ...

　　 return fork(cluster.settings.exec, cluster.settings.args, {

　　    cwd: cluster.settings.cwd,

　　    env: workerEnv,

　　    serialization: cluster.settings.serialization,

　　    silent: cluster.settings.silent,

　　    windowsHide: cluster.settings.windowsHide,

　　    execArgv: execArgv,

　　    stdio: cluster.settings.stdio,

　　    gid: cluster.settings.gid,

　　    uid: cluster.settings.uid,

　　 })

　　}

　　可以看到，该方法主要是通过 fork 启动了一个子进程来执行我们的 index.js，且启动子进程的时候设置了环境变量 NODE_UNIQUE_ID，这样 index.js 中 require('cluster') 的时候，引入的就是 internal/cluster/child.js 模块了。

　　worker.process.on('internalMessage', internal(worker, onmessage))：监听子进程传递过来的消息并处理。

　　接下来就进入了子进程的逻辑：

　　前面说了，此时引入的是 internal/cluster/child.js 模块，我们先跳过，继续往下，执行 server.listen(9999) 时实际上是调用了 Server 上的方法：

　　// lib/net.js

　　Server.prototype.listen = function (...args) {

　　 ...

　　      listenInCluster(

　　        this,

　　        null,

　　        options.port | 0,

　　        4,

　　        backlog,

　　        undefined,

　　        options.exclusive

　　      );

　　}

　　可以看到，最终是调用了 listenInCluster：

　　// lib/net.js

　　function listenInCluster(

　　 server,

　　 address,

　　 port,

　　 addressType,

　　 backlog,

　　 fd,

　　 exclusive,

　　 flags,

　　 options

　　) {

　　 exclusive = !!exclusive

　　 if (cluster === undefined) cluster = require('cluster')

　　 if (cluster.isPrimary || exclusive) {

　　    // Will create a new handle

　　    // _listen2 sets up the listened handle, it is still named like this

　　    // to avoid breaking code that wraps this method

　　    server._listen2(address, port, addressType, backlog, fd, flags)

　　    return

　　 }

　　 const serverQuery = {

　　    address: address,

　　    port: port,

　　    addressType: addressType,

　　    fd: fd,

　　    flags,

　　    backlog,

　　    ...options,

　　 }

　　 // Get the primary's server handle, and listen on it

　　 cluster._getServer(server, serverQuery, listenOnPrimaryHandle)

　　 function listenOnPrimaryHandle(err, handle) {

　　    err = checkBindError(err, port, handle)

　　    if (err) {

　　      const ex = exceptionWithHostPort(err, 'bind', address, port)

　　      return server.emit('error', ex)

　　    }

　　    // Reuse primary's server handle

　　    server._handle = handle

　　    // _listen2 sets up the listened handle, it is still named like this

　　    // to avoid breaking code that wraps this method

　　    server._listen2(address, port, addressType, backlog, fd, flags)

　　 }

　　}

　　由于是在子进程中执行，所以最后会调用 cluster._getServer(server, serverQuery, listenOnPrimaryHandle)：

　　// lib/internal/cluster/child.js

　　// 这里的 cb 就是上面的 listenOnPrimaryHandle

　　cluster._getServer = function (obj, options, cb) {

　　 ...

　　 send(message, (reply, handle) => {

　　    debugger

　　    if (typeof obj._setServerData === 'function') obj._setServerData(reply.data)

　　    if (handle) {

　　      // Shared listen socket

　　      shared(reply, {handle, indexesKey, index}, cb)

　　    } else {

　　      // Round-robin.

　　      rr(reply, {indexesKey, index}, cb)

　　    }

　　 })

　　 ...

　　}

　　该函数最终会向父进程发送 queryServer 的消息，父进程处理完后会调用回调函数，回调函数中会调用 cb 即 listenOnPrimaryHandle。看来，listen 的逻辑是在父进程中进行的了。

　　接下来进入父进程：

　　父进程收到 queryServer 的消息后，最终会调用 queryServer 这个方法：

　　// lib/internal/cluster/primary.js

　　function queryServer(worker, message) {

　　 // Stop processing if worker already disconnecting

　　 if (worker.exitedAfterDisconnect) return

　　 const key =

　　    `${message.address}:${message.port}:${message.addressType}:` +

　　    `${message.fd}:${message.index}`

　　 let handle = handles.get(key)

　　 if (handle === undefined) {

　　    let address = message.address

　　    // Find shortest path for unix sockets because of the ~100 byte limit

　　    if (

　　      message.port < 0 &&

　　      typeof address === 'string' &&

　　      process.platform !== 'win32'

　　    ) {

　　      address = path.relative(process.cwd(), address)

　　      if (message.address.length < address.length) address = message.address

　　    }

　　    // UDP is exempt from round-robin connection balancing for what should

　　    // be obvious reasons: it's connectionless. There is nothing to send to

　　    // the workers except raw datagrams and that's pointless.

　　    if (

　　      schedulingPolicy !== SCHED_RR ||

　　      message.addressType === 'udp4' ||

　　      message.addressType === 'udp6'

　　    ) {

　　      handle = new SharedHandle(key, address, message)

　　    } else {

　　      handle = new RoundRobinHandle(key, address, message)

　　    }

　　    handles.set(key, handle)

　　 }

　　 ...

　　}

　　可以看到，这里主要是对 handle 的处理，这里的 handle 指的是调度策略，分为 SharedHandle 和 RoundRobinHandle，分别对应抢占式和轮询两种策略（文章最后补充部分有关于两者对比的例子）。

　　Node.js 中默认是 RoundRobinHandle 策略，可通过环境变量 NODE_CLUSTER_SCHED_POLICY 来修改，取值可以为 none（SharedHandle）或 rr（RoundRobinHandle）。

　　SharedHandle

　　首先，我们来看一下 SharedHandle，由于我们这里是 TCP 协议，所以最后会通过 net._createServerHandle 创建一个 TCP 对象挂载在 handle 属性上（注意这里又有一个 handle，别搞混了）：

　　// lib/internal/cluster/shared_handle.js

　　function SharedHandle(key, address, {port, addressType, fd, flags}) {

　　 this.key = key

　　 this.workers = new SafeMap()

　　 this.handle = null

　　 this.errno = 0

　　 let rval

　　 if (addressType === 'udp4' || addressType === 'udp6')

　　    rval = dgram._createSocketHandle(address, port, addressType, fd, flags)

　　 else rval = net._createServerHandle(address, port, addressType, fd, flags)

　　 if (typeof rval === 'number') this.errno = rval

　　 else this.handle = rval

　　}

　　在 createServerHandle 中除了创建 TCP 对象外，还绑定了端口和地址：

　　// lib/net.js

　　function createServerHandle(address, port, addressType, fd, flags) {

　　 ...

　　 } else {

　　    handle = new TCP(TCPConstants.SERVER);

　　    isTCP = true;

　　 }

　　 if (address || port || isTCP) {

　　      ...

　　      err = handle.bind6(address, port, flags);

　　    } else {

　　      err = handle.bind(address, port);

　　    }

　　 }

　　 ...

　　 return handle;

　　}

　　然后，queryServer 中继续执行，会调用 add 方法，最终会将 handle 也就是 TCP 对象传递给子进程：

　　// lib/internal/cluster/primary.js

　　function queryServer(worker, message) {

　　 ...

　　 if (!handle.data) handle.data = message.data

　　 // Set custom server data

　　 handle.add(worker, (errno, reply, handle) => {

　　    const {data} = handles.get(key)

　　    if (errno) handles.delete(key) // Gives other workers a chance to retry.

　　    send(

　　      worker,

　　      {

　　        errno,

　　        key,

　　        ack: message.seq,

　　        data,

　　        ...reply,

　　      },

　　      handle // TCP 对象

　　    )

　　 })

　　 ...

　　}

　　之后进入子进程：

　　子进程收到父进程对于 queryServer 的回复后，会调用 shared：

　　// lib/internal/cluster/child.js

　　// `obj` is a net#Server or a dgram#Socket object.

　　cluster._getServer = function (obj, options, cb) {

　　 ...

　　 send(message, (reply, handle) => {

　　    if (typeof obj._setServerData === 'function') obj._setServerData(reply.data)

　　    if (handle) {

　　      // Shared listen socket

　　      shared(reply, {handle, indexesKey, index}, cb)

　　    } else {

　　      // Round-robin.

　　      rr(reply, {indexesKey, index}, cb) // cb 是 listenOnPrimaryHandle

　　    }

　　 })

　　 ...

　　}

　　shared 中最后会调用 cb 也就是 listenOnPrimaryHandle：

　　// lib/net.js

　　function listenOnPrimaryHandle(err, handle) {

　　 err = checkBindError(err, port, handle)

　　 if (err) {

　　    const ex = exceptionWithHostPort(err, 'bind', address, port)

　　    return server.emit('error', ex)

　　 }

　　 // Reuse primary's server handle 这里的 server 是 index.js 中 net.createServer 返回的那个对象

　　 server._handle = handle

　　 // _listen2 sets up the listened handle, it is still named like this

　　 // to avoid breaking code that wraps this method

　　 server._listen2(address, port, addressType, backlog, fd, flags)

　　}

　　这里会把 handle 赋值给 server._handle，这里的 server 是 index.js 中 net.createServer 返回的那个对象，并调用 server._listen2，也就是 setupListenHandle：

　　// lib/net.js

　　function setupListenHandle(address, port, addressType, backlog, fd, flags) {

　　 debug('setupListenHandle', address, port, addressType, backlog, fd)

　　 // If there is not yet a handle, we need to create one and bind.

　　 // In the case of a server sent via IPC, we don't need to do this.

　　 if (this._handle) {

　　    debug('setupListenHandle: have a handle already')

　　 } else {

　　    ...

　　 }

　　 this[async_id_symbol] = getNewAsyncId(this._handle)

　　 this._handle.onconnection = onconnection

　　 this._handle[owner_symbol] = this

　　 // Use a backlog of 512 entries. We pass 511 to the listen() call because

　　 // the kernel does: backlogsize = roundup_pow_of_two(backlogsize + 1);

　　 // which will thus give us a backlog of 512 entries.

　　 const err = this._handle.listen(backlog || 511)

　　 if (err) {

　　    const ex = uvExceptionWithHostPort(err, 'listen', address, port)

　　    this._handle.close()

　　    this._handle = null

　　    defaultTriggerAsyncIdScope(

　　      this[async_id_symbol],

　　      process.nextTick,

　　      emitErrorNT,

　　      this,

　　      ex

　　    )

　　    return

　　 }

　　}

　　首先会执行 this._handle.onconnection = onconnection，由于客户端请求过来时会调用 this._handle（也就是 TCP 对象）上的 onconnection 方法，也就是会执行lib/net.js 中的 onconnection 方法建立连接，之后就可以通信了。为了控制篇幅，该方法就不继续往下了。

　　然后调用 listen 监听，注意这里参数 backlog 跟之前不同，不是表示端口，而是表示在拒绝连接之前，操作系统可以挂起的最大连接数量，也就是连接请求的排队数量。我们平时遇到的 listen EADDRINUSE: address already in use 错误就是因为这行代码返回了非 0 的错误。

　　如果还有其他子进程，也会同样走一遍上述的步骤，不同之处是在主进程中 queryServer 时，由于已经有 handle 了，不需要再重新创建了：

　　function queryServer(worker, message) {

　　 debugger;

　　 // Stop processing if worker already disconnecting

　　 if (worker.exitedAfterDisconnect) return;

　　 const key =

　　    `${message.address}:${message.port}:${message.addressType}:` +

　　    `${message.fd}:${message.index}`;

　　 let handle = handles.get(key);

　　 ...

　　}

（编辑：厦门网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!