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

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)

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 runningWorker 11852 startedWorker 11854 startedWorker 11853 startedWorker 11855 startedWorker 11857 startedWorker 11858 startedWorker 11856 startedWorker 11859 started

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

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

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

NODE_CLUSTER_SCHED_POLICY='rr'NODE_CLUSTER_SCHED_POLICY='none'

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

RoundRobinHandle.prototype.distribute = function(err, handle) { this.handles.push(handle); const worker = this.free.shift(); if (worker) { this.handoff(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); });};