Node.js中如何使用Cluster模块

Node.js中如何使用Cluster模块，很多新手对此不是很清楚，为了帮助大家解决这个难题，下面小编将为大家详细讲解，有这方面需求的人可以来学习下，希望你能有所收获。

专业领域包括成都网站建设、成都网站设计、商城网站建设、微信营销、系统平台开发， 与其他网站设计及系统开发公司不同，创新互联公司的整合解决方案结合了帮做网络品牌建设经验和互联网整合营销的理念，并将策略和执行紧密结合，为客户提供全网互联网整合方案。

1.为什么我的应用代码中明明有app.listen(port);，但cluter模块在多次fork这份代码时，却没有报端口已被占用？

2.Master是如何将接收的请求传递至worker中进行处理然后响应的？

带着这些疑问我们开始往下看

TIPS：

本文编写于2019年12月8日，是最新版本的Node.js源码

Cluster源码解析：

•  入口 ： 

const childOrMaster = 'NODE_UNIQUE_ID' in process.env ? 'child' : 'master';  module.exports = require(`internal/cluster/${childOrMaster}`);

•  分析

会根据一个当前的Node_UNIQUE_ID（后面会讲）是否在环境变量中判断是子进程还是主进程，然后引用不同的js代码

NODE_UNIQUE_ID是一个唯一标示，Node.js的Cluster多进程模式，采用默认的调度算法是round-robin,其实就是轮询.官方解释是实践效率非常高，稳定

之前的问题一： 为什么我的应用代码中明明有app.listen(port);，但cluter模块在多次fork这份代码时，却没有报端口已被占用？

我在Node.js的官网找到了答案：

原来所有的net.Socket都被设置了SO_REUSEADDR

这个SO_REUSEADDR到底是什么呢？

为什么需要 SO_REUSEADDR 参数？

服务端主动断开连接以后，需要等 2 个 MSL 以后才最终释放这个连接，重启以后要绑定同一个端口，默认情况下，操作系统的实现都会阻止新的监听套接字绑定到这个端口上。

我们都知道 TCP 连接由四元组唯一确定。形式如下

{local-ip-address:local-port , foreign-ip-address:foreign-port}

一个典型的例子如下图

TCP 要求这样的四元组必须是唯一的，但大多数操作系统的实现要求更加严格，只要还有连接在使用这个本地端口，则本地端口不能被重用（bind 调用失败）

启用 SO_REUSEADDR 套接字选项可以解除这个限制，默认情况下这个值都为 0，表示关闭。在 Java 中，reuseAddress 不同的 JVM 有不同的实现，在我本机上，这个值默认为 1 允许端口重用。但是为了保险起见，写 TCP、HTTP 服务一定要主动设置这个参数为 1。

目前常见的网络编程模型就是多进程或多线程，根据accpet的位置，分为如下场景

2种场景

(1) 单进程或线程创建socket，并进行listen和accept，接收到连接后创建进程和线程处理连接

(2) 单进程或线程创建socket，并进行listen，预先创建好多个工作进程或线程accept()在同一个服务器套接字

这两种模型解充分发挥了多核CPU的优势,虽然可以做到线程和CPU核绑定，但都会存在：

1.单一listener工作进程或线程在高速的连接接入处理时会成为瓶颈

2.多个线程之间竞争获取服务套接字

3.缓存行跳跃

4.很难做到CPU之间的负载均衡

5.随着核数的扩展，性能并没有随着提升

6.SO_REUSEPORT解决了什么问题

7.SO_REUSEPORT支持多个进程或者线程绑定到同一端口，提高服务器程序的性能

解决的问题：

1.允许多个套接字 bind()/listen() 同一个TCP/UDP端口

2.每一个线程拥有自己的服务器套接字

3.在服务器套接字上没有了锁的竞争

4.内核层面实现负载均衡

5.安全层面，监听同一个端口的套接字只能位于同一个用户下面

其核心的实现主要有三点：

1.扩展 socket option，增加 SO_REUSEPORT 选项，用来设置 reuseport

2.修改 bind 系统调用实现，以便支持可以绑定到相同的 IP 和端口

3.修改处理新建连接的实现，查找 listener 的时候，能够支持在监听相同 IP 4.和端口的多个 sock 之间均衡选择。

5.有了SO_RESUEPORT后，每个进程可以自己创建socket、bind、listen、accept相同的地址和端口，各自是独立平等的

让多进程监听同一个端口，各个进程中accept socket fd不一样，有新连接建立时，内核只会唤醒一个进程来accept，并且保证唤醒的均衡性。

总结：原来端口被复用是因为设置了SO_REUSEADDR，当然不止这一点，下面会继续描述

回到源码第一行

NODE_UNIQUE_ID是什么？

下面给出介绍：

function createWorkerProcess(id, env) {    // ...    workerEnv.NODE_UNIQUE_ID = '' + id;    // ...    return fork(cluster.settings.exec, cluster.settings.args, {      env: workerEnv,      silent: cluster.settings.silent,      execArgv: execArgv,      gid: cluster.settings.gid,      uid: cluster.settings.uid    });  }

原来，创建子进程的时候，给了每个进程一个唯一的自增标示ID

随后Node.js在初始化时，会根据该环境变量，来判断该进程是否为cluster模块fork出的工作进程，若是，则执行workerInit()函数来初始化环境，否则执行masterInit()函数

就是这行入口的代码～

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

接下来我们需要看一下net模块的listen函数源码：

// lib/net.js  // ...  function listen(self, address, port, addressType, backlog, fd, exclusive) {    exclusive = !!exclusive;    if (!cluster) cluster = require('cluster');    if (cluster.isMaster || exclusive) {      self._listen2(address, port, addressType, backlog, fd);      return;    }    cluster._getServer(self, {      address: address,      port: port,      addressType: addressType,      fd: fd,      flags: 0    }, cb);    function cb(err, handle) {      // ...      self._handle = handle;      self._listen2(address, port, addressType, backlog, fd);    }  }

仔细一看，原来listen函数会根据是不是主进程做不同的操作！

上面有提到SO_REUSEADDR选项，在主进程调用的_listen2中就有设置。

子进程初始化的每个workerinit函数中，也有cluster._getServer这个方法，

你可能已经猜到，问题一的答案，就在这个cluster._getServer函数的代码中。它主要干了两件事：

•  向master进程注册该worker，若master进程是第一次接收到监听此端口/描述符下的worker，则起一个内部TCP服务器，来承担监听该端口/描述符的职责，随后在master中记录下该worker。

•  Hack掉worker进程中的net.Server实例的listen方法里监听端口/描述符的部分，使其不再承担该职责。

对于第一件事，由于master在接收，传递请求给worker时，会符合一定的负载均衡规则（在非Windows平台下默认为轮询），这些逻辑被封装在RoundRobinHandle类中。故，初始化内部TCP服务器等操作也在此处：

// lib/cluster.js  // ...  function RoundRobinHandle(key, address, port, addressType, backlog, fd) {    // ...    this.handles = [];    this.handle = null;    this.server = net.createServer(assert.fail);    if (fd >= 0)      this.server.listen({ fd: fd });    else if (port >= 0)      this.server.listen(port, address);    else      this.server.listen(address);  // UNIX socket path.    /// ...  }

在子进程中：

function listen(backlog) {      return 0;    }    function close() {      // ...    }    function ref() {}    function unref() {}    var handle = {      close: close,      listen: listen,      ref: ref,      unref: unref,    }

由于net.Server实例的listen方法，最终会调用自身_handle属性下listen方法来完成监听动作，故在代码中修改之：此时的listen方法已经被hack ，每次调用只能发挥return 0 ,并不会监听端口

// lib/net.js  // ...  function listen(self, address, port, addressType, backlog, fd, exclusive) {    // ...    if (cluster.isMaster || exclusive) {      self._listen2(address, port, addressType, backlog, fd);      return; // 仅在worker环境下改变    }    cluster._getServer(self, {      address: address,      port: port,      addressType: addressType,      fd: fd,      flags: 0    }, cb);    function cb(err, handle) {      // ...      self._handle = handle;      // ...    }  }

这里可以看到，传入的回调函数中的handle，已经把listen方法重新定义，返回0，那么等子进程调用listen方法时候，也是返回0，并不会去监听端口，至此，焕然大悟，原来是这样，真正监听端口的始终只有主进程!

上面通过将近3000字讲解，把端口复用这个问题讲清楚了，下面把负载均衡这块也讲清楚。然后再讲PM2的原理实现,其实不过是对cluster模式进行了封装，多了很多功能而已～

首先画了一个流程图

核心实现源码：

function RoundRobinHandle(key, address, port, addressType, backlog, fd) {    // ...    this.server = net.createServer(assert.fail);    // ...    var self = this;    this.server.once('listening', function() {      // ...      selfself.handle.onconnection = self.distribute.bind(self);    });  }  RoundRobinHandle.prototype.distribute = function(err, handle) {    this.handles.push(handle);    var worker = this.free.shift();    if (worker) this.handoff(worker);  };  RoundRobinHandle.prototype.handoff = function(worker) {    // ...    var message = { act: 'newconn', key: this.key };    var self = this;    sendHelper(worker.process, message, handle, function(reply) {      // ...    });

解析

定义好handle对象中的onconnection方法

触发事件时，取出一个子进程通知，传入句柄

子进程接受到消息和句柄后，做相应的业务处理：

 var accepted = server !== undefined;    // ...    if (accepted) server.onconnection(0, handle);// lib/cluster.js  // ...  // 该方法会在Node.js初始化时由 src/node.js 调用  cluster._setupWorker = function() {    // ...    process.on('internalMessage', internal(worker, onmessage));      // ...    function onmessage(message, handle) {      if (message.act === 'newconn')        onconnection(message, handle);      // ...    }  };  function onconnection(message, handle) {    // ...  }

总结下来，负载均衡大概流程：

1.所有请求先同一经过内部TCP服务器，真正监听端口的只有主进程。

2.在内部TCP服务器的请求处理逻辑中，有负载均衡地挑选出一个worker进程，将其发送一个newconn内部消息，随消息发送客户端句柄。

3.Worker进程接收到此内部消息，根据客户端句柄创建net.Socket实例，执行具体业务逻辑，返回。

至此，Cluster多进程模式，负载均衡讲解完毕，下面讲PM2的实现原理，它是基于Cluster模式的封装

PM2的使用:

npm i pm2 -g   pm2 start app.js   pm2 ls

这样就可以启动你的Node.js服务，并且根据你的电脑CPU个数去启动相应的进程数，监听到错误事件，自带重启子进程，即使更新了代码，需要热更新，也会逐个替换，号称永动机。

它的功能：

1.内建负载均衡（使用Node cluster 集群模块）

2.后台运行

3.0秒停机重载，我理解大概意思是维护升级的时候不需要停机.

4.具有Ubuntu和CentOS 的启动脚本

5.停止不稳定的进程（避免无限循环）

6.控制台检测

7.提供 HTTP API

8.远程控制和实时的接口API ( Nodejs 模块,允许和PM2进程管理器交互 )

先来一张PM2的架构图：

pm2包括 Satan进程、God Deamon守护进程、进程间的远程调用rpc、cluster等几个概念

如果不知道点西方文化，还真搞不清他的文件名为啥是 Satan 和 God：

撒旦（Satan），主要指《圣经》中的堕天使（也称堕天使撒旦），被看作与上帝的力量相对的邪恶、黑暗之源，是God的对立面。

1.Satan.js提供了程序的退出、杀死等方法，因此它是魔鬼；God.js 负责维护进程的正常运行，当有异常退出时能保证重启，所以它是上帝。作者这么命名，我只能说一句：oh my god。

God进程启动后一直运行，它相当于cluster中的Master进程，守护者worker进程的正常运行。

2.rpc（Remote Procedure Call Protocol）是指远程过程调用，也就是说两台服务器A，B，一个应用部署在A服务器上，想要调用B服务器上应用提供的函数/方法，由于不在一个内存空间，不能直接调用，需要通过网络来表达调用的语义和传达调用的数据。同一机器不同进程间的方法调用也属于rpc的作用范畴。

3.代码中采用了axon-rpc 和 axon 两个库，基本原理是提供服务的server绑定到一个域名和端口下，调用服务的client连接端口实现rpc连接。后续新版本采用了pm2-axon-rpc 和 pm2-axon两个库，绑定的方法也由端口变成.sock文件，因为采用port可能会和现有进程的端口产生冲突。

执行流程

程序的执行流程图如下：

每次命令行的输入都会执行一次satan程序。如果God进程不在运行，首先需要启动God进程。然后根据指令，satan通过rpc调用God中对应的方法执行相应的逻辑。

以 pm2 start app.js -i 4为例，God在初次执行时会配置cluster，同时监听cluster中的事件：

// 配置cluster  cluster.setupMaster({    exec : path.resolve(path.dirname(module.filename), 'ProcessContainer.js')  });  // 监听cluster事件  (function initEngine() {    cluster.on('online', function(clu) {      // worker进程在执行      God.clusters_db[clu.pm_id].status = 'online';    });    // 命令行中 kill pid 会触发exit事件，process.kill不会触发exit    cluster.on('exit', function(clu, code, signal) {      // 重启进程 如果重启次数过于频繁直接标注为stopped      God.clusters_db[clu.pm_id].status = 'starting';      // 逻辑      ...    });  })();

在God启动后， 会建立Satan和God的rpc链接，然后调用prepare方法。prepare方法会调用cluster.fork，完成集群的启动

God.prepare = function(opts, cb) {    ...    return execute(opts, cb);  };  function execute(env, cb) {    ...    var clu = cluster.fork(env);    ...    God.clusters_db[id] = clu;    clu.once('online', function() {      God.clusters_db[id].status = 'online';      if (cb) return cb(null, clu);      return true;    });    return clu;  }

PM2的功能目前已经特别多了，源码阅读非常耗时，但是可以猜测到一些功能的实现：

例如

如何检测子进程是否处于正常活跃状态？

采用心跳检测

每隔数秒向子进程发送心跳包，子进程如果不回复，那么调用kill杀死这个进程  然后再重新cluster.fork()一个新的进程

子进程发出异常报错，如何保证一直有一定数量子进程？

子进程可以监听到错误事件，这时候可以发送消息给主进程，请求杀死自己  并且主进程此时重新调用cluster.fork一个新的子进程

看完上述内容是否对您有帮助呢？如果还想对相关知识有进一步的了解或阅读更多相关文章，请关注创新互联行业资讯频道，感谢您对创新互联的支持。