伏龙阁资源网

详解通过源码解析Node.js中cluster模块的主要功能实现
站长资源 2024/11/27 佚名
2 0

众所周知，Node.js中的JavaScript代码执行在单线程中，非常脆弱，一旦出现了未捕获的异常，那么整个应用就会崩溃。这在许多场景下，尤其是web应用中，是无法忍受的。通常的解决方案，便是使用Node.js中自带的cluster模块，以master-worker模式启动多个应用实例。然而大家在享受cluster模块带来的福祉的同时，不少人也开始好奇：

  1. 为什么我的应用代码中明明有app.listen(port);，但cluter模块在多次fork这份代码时，却没有报端口已被占用？
  2. Master是如何将接收的请求传递至worker中进行处理然后响应的？

让我们从Node.js项目的lib/cluster.js中的代码里，来一勘究竟。

问题一

为了得到这个问题的解答，我们先从worker进程的初始化看起，master进程在fork工作进程时，会为其附上环境变量NODE_UNIQUE_ID，是一个从零开始的递增数：

// lib/cluster.js
// ...

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

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

在workerInit()函数中，定义了cluster._getServer方法，这个方法在任何net.Server实例的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);
 }
}

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

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

 /// ...
}

对于第二件事，由于net.Server实例的listen方法，最终会调用自身_handle属性下listen方法来完成监听动作，故在代码中修改之：

// lib/cluster.js
// ...

function rr(message, cb) {
 // ...
 // 此处的listen函数不再做任何监听动作
 function listen(backlog) {
  return 0;
 }

 function close() {
  // ...
 }
 function ref() {}
 function unref() {}

 var handle = {
  close: close,
  listen: listen,
  ref: ref,
  unref: unref,
 };
 // ...
 handles[key] = handle;
 cb(0, handle); // 传入这个cb中的handle将会被赋值给net.Server实例中的_handle属性
}

// 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;
  // ...
 }
}

至此，第一个问题便已豁然开朗了，总结下：

  1. 端口仅由master进程中的内部TCP服务器监听了一次。
  2. 不会出现端口被重复监听报错，是由于，worker进程中，最后执行监听端口操作的方法，已被cluster模块主动hack。

问题二

解决了问题一，问题二的解决就明朗轻松许多了。通过问题一我们已得知，监听端口的是master进程中创建的内部TCP服务器，所以第二个问题的解决，着手点就是该内部TCP服务器接手连接时，执行的操作。Cluster模块的做法是，监听该内部TCP服务器的connection事件，在监听器函数里，有负载均衡地挑选出一个worker，向其发送newconn内部消息（消息体对象中包含cmd: 'NODE_CLUSTER'属性）以及一个客户端句柄（即connection事件处理函数的第二个参数），相关代码如下：

// lib/cluster.js
// ...

function RoundRobinHandle(key, address, port, addressType, backlog, fd) {
 // ...
 this.server = net.createServer(assert.fail);
 // ...

 var self = this;
 this.server.once('listening', function() {
  // ...
  self.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) {
  // ...
 });
};

Worker进程在接收到了newconn内部消息后，根据传递过来的句柄，调用实际的业务逻辑处理并返回：

// 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) {
 // ...
 var accepted = server !== undefined;
 // ...
 if (accepted) server.onconnection(0, handle);
}

至此，问题二也得到了解决，也总结一下：

  1. 所有请求先同一经过内部TCP服务器。
  2. 在内部TCP服务器的请求处理逻辑中，有负载均衡地挑选出一个worker进程，将其发送一个newconn内部消息，随消息发送客户端句柄。
  3. Worker进程接收到此内部消息，根据客户端句柄创建net.Socket实例，执行具体业务逻辑，返回。

最后

Node.js中的cluster模块除了上述提到的功能外，其实还提供了非常丰富的API供master和worker进程之前通信，对于不同的操作系统平台，也提供了不同的默认行为。本文仅挑选了一条功能线进行了分析阐述。如果大家有闲，非常推荐完整领略一下cluster模块的代码实现。

参考：

https://github.com/nodejs/node/blob/master/lib/cluster.js
https://github.com/nodejs/node/blob/master/lib/net.js

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。

广告合作：本站广告合作请联系QQ：858582 申请时备注：广告合作（否则不回）
免责声明：本站资源来自互联网收集,仅供用于学习和交流,请遵循相关法律法规,本站一切资源不代表本站立场,如有侵权、后门、不妥请联系本站删除！
www.wwsws.com 伏龙阁资源网
39,976影音资源
44,792技术资源
21,817软件资源
651,128站长资源

更新日志

友情链接

杰晶网络 DDR爱好者之家 桃源资源网 杰网资源 富贵资源网 南强小屋 铁雪资源网 幽灵资源网 万梅资源网 狼山资源网 白云岛资源网 昆仑资源网 相思资源网 明霞山资源网 内蒙古资源网 黑松山资源网 茶园资源网 饿虎岗资源网 大旗谷资源网 常春岛资源网 岱庙资源网 兴国资源网 快活林资源网 蝙蝠岛资源网 帝王谷资源网 白云城资源网 伏龙阁资源网 清风细雨楼 天枫庄资源网 圆月山庄资源网 无争山庄资源网 神水资源网 移花宫资源网 神剑山庄资源网 无为清净楼资源网 金钱帮资源网 丐帮资源网 华山资源网 极乐门资源网 小李飞刀资源网 凤求凰客栈 风云阁资源网 金狮镖局 鸳鸯亭资源网 千金楼资源网 更多链接