Node.js中的cluster模块深入解读

预备知识

在如今机器的CPU都是多核的背景下，Node的单线程设计已经没法更充分的"压榨"机器性能了。所以从v0.8开始，Node新增了一个内置模块——“cluster”，故名思议，它可以通过一个父进程管理一坨子进程的方式来实现集群的功能。

学习cluster之前，需要了解process相关的知识，如果不了解的话建议先阅读process模块、child_process模块。

cluster借助child_process模块的fork()方法来创建子进程，通过fork方式创建的子进程与父进程之间建立了IPC通道，支持双向通信。

cluster模块最早出现在node.js v0.8版本中

为什么会存在cluster模块？

Node.js是单线程的，那么如果希望利用服务器的多核的资源的话，就应该多创建几个进程，由多个进程共同提供服务。如果直接采用下列方式启动多个服务的话，会提示端口占用。

const http = require('http');
http.createServer((req, res) => {
 res.writeHead(200);
 res.end('hello world\n');
}).listen(8000);

// 启动第一个服务 node index.js &
// 启动第二个服务 node index.js &

 throw er; // Unhandled 'error' event
 ^

Error: listen EADDRINUSE :::8000
 at Server.setupListenHandle [as _listen2] (net.js:1330:14)
 at listenInCluster (net.js:1378:12)
 at Server.listen (net.js:1465:7)
 at Object.<anonymous> (/Users/xiji/workspace/learn/node-basic/cluster/simple.js:5:4)
 at Module._compile (internal/modules/cjs/loader.js:702:30)
 at Object.Module._extensions..js (internal/modules/cjs/loader.js:713:10)
 at Module.load (internal/modules/cjs/loader.js:612:32)
 at tryModuleLoad (internal/modules/cjs/loader.js:551:12)
 at Function.Module._load (internal/modules/cjs/loader.js:543:3)
 at Function.Module.runMain (internal/modules/cjs/loader.js:744:10)

如果改用cluster的话就没有问题

const cluster = require('cluster');
const http = require('http');
const numCPUs = require('os').cpus().length;

if (cluster.isMaster) {
 console.log(`Master ${process.pid} is running`);

 // Fork workers.
 for (let i = 0; i < numCPUs; i++) {
 cluster.fork();
 }

 cluster.on('exit', (worker, code, signal) => {
 console.log(`worker ${worker.process.pid} died`);
 });
} else {
 // Workers can share any TCP connection
 // In this case it is an HTTP server
 http.createServer((req, res) => {
 res.writeHead(200);
 res.end('hello world\n');
 }).listen(8000);

 console.log(`Worker ${process.pid} started`);
}

// node index.js 执行完启动了一个主进程和8个子进程(子进程数与cpu核数相一致)
Master 11851 is running
Worker 11852 started
Worker 11854 started
Worker 11853 started
Worker 11855 started
Worker 11857 started
Worker 11858 started
Worker 11856 started
Worker 11859 started

cluster是如何实现多进程共享端口的？

cluster创建的进程分两种，父进程和子进程，父进程只有一个，子进程有多个（一般根据cpu核数创建）

• 父进程负责监听端口接受请求，然后分发请求。
• 子进程负责请求的处理。

有三个问题需要回答：

• 子进程为何调用listen不会进行端口绑定
• 父进程何时创建的TCP Server
• 父进程是如何完成分发的

子进程为何调用listen不会绑定端口？

net.js源码中的listen方法通过listenInCluster方法来区分是父进程还是子进程，不同进程的差异在listenInCluster方法中体现

function listenInCluster(server, address, port, addressType, backlog, fd, excluseive) {

 if (cluster.isMaster || exclusive) {
 server._listen2(address, port, addressType, backlog, fd);
 return;
 }

 const serverQuery = { address: address ......};

 cluster._getServer(server, serverQuery, listenOnMasterHandle);

 function listenOnMasterHandle(err, handle) {
 server._handle = handle;
 server._listen2(address, port, addressType, backlog, fd);
 }
}

上面是精简过的代码，当子进程调用listen方法时，会先执行_getServer，然后通过callback的形式指定server._handle的值，之后再调用_listen2方法。

cluster._getServer = function(obj, options, cb) {
 ...
 const message = util._extend({
 act: 'queryServer',
 index: indexes[indexesKey],
 data: null
 }, options);

 message.address = address;

 send(message, (reply, handle) => {
 if (handle)
 shared(reply, handle, indexesKey, cb); // Shared listen socket.
 else
 rr(reply, indexesKey, cb); // Round-robin.
 });
 ...
};

_getServer方法会向主进程发送queryServer的message，父进程执行完会调用回调函数，根据是否返回handle来区分是调用shared方法还是rr方法，这里其实是会调用rr方法。而rr方法的主要作用就是伪造了TCPWrapper来调用net的listenOnMasterHandle回调函数

function rr(message, indexesKey, cb) {

 var key = message.key;

 function listen(backlog) {
 return 0;
 }

 function close() {
 if (key === undefined)
 return;

 send({ act: 'close', key });
 delete handles[key];
 delete indexes[indexesKey];
 key = undefined;
 }

 function getsockname(out) {
 if (key)
 util._extend(out, message.sockname);

 return 0;
 }

 const handle = { close, listen, ref: noop, unref: noop };
 handles[key] = handle;
 cb(0, handle);
}

由于子进程的server拿到的是围绕的TCPWrapper，当调用listen方法时并不会执行任何操作，所以在子进程中调用listen方法并不会绑定端口，因而也并不会报错。

父进程何时创建的TCP Server

在子进程发送给父进程的queryServer message时，父进程会检测是否创建了TCP Server，如果没有的话就会创建TCP Server并绑定端口，然后再把子进程记录下来，方便后续的用户请求worker分发。

父进程是如何完成分发的

父进程由于绑定了端口号，所以可以捕获连接请求，父进程的onconnection方法会被触发，onconnection方法触发时会传递TCP对象参数，由于之前父进程记录了所有的worker，所以父进程可以选择要处理请求的worker，然后通过向worker发送act为newconn的消息，并传递TCP对象，子进程监听到消息后，对传递过来的TCP对象进行封装，封装成socket，然后触发connection事件。这样就实现了子进程虽然不监听端口，但是依然可以处理用户请求的目的。

cluster如何实现负载均衡

负载均衡直接依赖cluster的请求调度策略，在v6.0版本之前，cluster的调用策略采用的是cluster.SCHED_NONE（依赖于操作系统），SCHED_NODE理论上来说性能最好（Ferando Micalli写过一篇Node.js 6.0版本的cluster和iptables以及nginx性能对比的文章）但是从实际角度发现，在请求调度方面会出现不太均匀的情况（可能出现8个子进程中的其中2到3个处理了70%的连接请求）。因此在6.0版本中Node.js增加了cluster.SCHED_RR（round-robin），目前已成为默认的调度策略（除了windows环境）

可以通过设置NODE_CLUSTER_SCHED_POLICY环境变量来修改调度策略

NODE_CLUSTER_SCHED_POLICY='rr'
NODE_CLUSTER_SCHED_POLICY='none'

或者设置cluster的schedulingPolicy属性

cluster.schedulingPolicy = cluster.SCHED_NONE;
cluster.schedulingPolicy = cluster.SCHED_RR;

Node.js实现round-robin

Node.js内部维护了两个队列：

• free队列记录当前可用的worker
• handles队列记录需要处理的TCP请求

当新请求到达的时候父进程将请求暂存handles队列，从free队列中出队一个worker，进入worker处理(handoff)阶段，关键逻辑实现如下：

RoundRobinHandle.prototype.distribute = function(err, handle) {
 this.handles.push(handle);
 const worker = this.free.shift();

 if (worker) {
 this.handoff(worker);
 }
};

worker处理阶段首先从handles队列出队一个请求，然后通过进程通信的方式通知子worker进行请求处理，当worker接收到通信消息后发送ack信息，继续响应handles队列中的请求任务，当worker无法接受请求时，父进程负责重新调度worker进行处理。关键逻辑如下：

RoundRobinHandle.prototype.handoff = function(worker) {
 const handle = this.handles.shift();
 if (handle === undefined) {
 this.free.push(worker); // Add to ready queue again.
 return;
 }

 const message = { act: 'newconn', key: this.key };
 sendHelper(worker.process, message, handle, (reply) => {
 if (reply.accepted)
 handle.close();
 else
 this.distribute(0, handle); // Worker is shutting down. Send to another.
 this.handoff(worker);
 });
};

注意：主进程与子进程之间建立了IPC，因此主进程与子进程之间可以通信，但是各个子进程之间是相互独立的（无法通信）

参考资料

https://medium.com/@fermads/node-js-process-load-balancing-comparing-cluster-iptables-and-nginx-6746aaf38272

总结

以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作具有一定的参考学习价值，如果有疑问大家可以留言交流，谢谢大家对我们的支持。