众所周知,node . js中的JavaScript代码执行在单线程中,非常脆弱,一旦出现了未捕获的异常,那么整个应用就会崩溃。这在许多场景下,尤其是网络应用中,是无法忍受的。通常的解决方案,便是使用node . js中自带的集群模块,以成为集大成模式启动多个应用实例。然而大家在享受集群模块带来的福祉的同时,不少人也开始好奇:
<李>为什么我的应用代码中明明有app.listen(港口);,但clut模块在多次叉这份代码时,却没有报端口已被占用?李
<李>主是如何将接收的请求传递至工人中进行处理然后响应的?李
让我们从node . js项目的lib/cluster.js中的代码里,来一勘究竟。
为了得到这个问题的解答,我们先从工人进程的初始化看起,主进程在叉工作进程时,会为其附上环境变量NODE_UNIQUE_ID,是一个从零开始的递增数:
//lib/cluster.js//?
函数createWorkerProcess (id、env) {//?
workerEnv。NODE_UNIQUE_ID=" +身份证;//?
返回叉(cluster.settings。exec cluster.settings。args, {
env: workerEnv,
沉默:cluster.settings.silent,
execArgv: execArgv,
gid: cluster.settings.gid,
uid: cluster.settings.uid
});
}
之前
随后node . js在初始化时,会根据该环境变量,来判断该进程是否为集群模块叉出的工作进程,若是,则执行workerInit()函数来初始化环境,否则执行masterInit()函数。
在workerInit()函数中,定义了cluster._getServer方法,这个方法在任何net.Server实例的听方法中,会被调用:
//lib/net.js//?
函数听(自我、地址、端口、addressType积压,fd,独家){
独家=! !独家;
如果集群(!)集群=要求(“集群”);
如果集群。isMaster | |独家){
自我。_listen2(地址、端口、addressType积压,fd);
返回;
}
集群。_getServer(自我,{
地址:地址,
端口:港口,
addressType: addressType,
fd: fd,
国旗:0
},cb);
函数cb(呃,处理){//?
自我。_handle=处理;
自我。_listen2(地址、端口、addressType积压,fd);
}
}
之前
你可能已经猜的到,问题一的答案,就在这个cluster._getServer函数的代码中。它主要干了两件事:
<李>向主人进程注册该工人,若主进程是第一次接收到监听此端口/描述符下的工人,则起一个内部TCP服务器,来承担监听该端口/描述符的职责,随后在主中记录下该工人。
<李>破解掉工人进程中的net.Server实例的听方法里监听端口/描述符的部分,使其不再承担该职责。
对于第一件事,由于大师在接收,传递请求给工人时,会符合一定的负载均衡规则(在非Windows平台下默认为轮询),这些逻辑被封装在RoundRobinHandle类中,故初始化内部TCP服务器等操作也在此处:
//lib/cluster.js//?
函数RoundRobinHandle(关键、地址、端口、addressType、积压、fd) {//?
这一点。处理=[];
这一点。处理=零;
这一点。服务器=net.createServer (assert.fail);
如果(fd祝辞=0)
this.server。听({fd: fd});
else if(端口在=0)
this.server。听(港口,地址);
其他的
this.server.listen(地址);//UNIX socket路径。///?
}
之前
对于第二件事,由于net.Server实例的听方法,最终会调用自身_handle属性下听方法来完成监听动作,故在代码中修改之:
//lib/cluster.js//?
函数rr(消息,cb) {//?/此处的听函数不再做任何监听动作
函数听(积压){
返回0;
}
函数close () {//?
}
函数参考(){}
函数unref () {}
var={处理
关闭:关闭,
听:听着,
裁判:裁判,
unref: unref,
}；//?
处理(例子)=处理;
cb(0,处理);//传入这个cb中的处理将会被赋值给net.Server实例中的_handle属性
}//lib/net.js//?
函数听(自我、地址、端口、addressType积压,fd,独家){//?
如果集群。isMaster | |独家){
自我。_listen2(地址、端口、addressType积压,fd);
返回;//仅在工作环境下改变
}
集群。_getServer(自我,{
地址:地址,
端口:港口,
addressType: addressType,
fd: fd,
国旗:0
},cb);
函数cb(呃,处理){//?
自我。_handle=处理;//?
}
}
