详解通过源码解析Node.js中cluster模块的主要功能实现

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

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

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

 /// ...
}

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

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

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