NodeJS Cluster模塊源碼學習

？

一段常見的示例代碼

const cluster = require('cluster');
const http = require('http');

if (cluster.isMaster) {
  // 根據cpu核心數出fork相同數量的子進程
} else {
  // 用http模塊建立server監聽某一個端口
}
複製代碼

引出以下問題：

1. cluster模塊如何區分子進程和主進程？

2. 代碼中沒有在主進程中建立服務器，那麼如何主進程如何承擔代理服務器的職責？

3. 多個子進程共同偵聽同一個端口爲何不會形成端口reuse error？

1. cluster模塊如何區分主進程/子進程

cluster.js - 源碼

const childOrMaster = 'NODE_UNIQUE_ID' in process.env ? 'child' : 'master';
module.exports = require(`internal/cluster/${childOrMaster}`);
複製代碼

結論: 判斷環境變量中是否含有NODE_UNIQUE_ID, 有則爲子進程，沒有則爲主進程

1.1 isMaster & isWorker

這樣的話, 在對應的文件中isMaster和isWorker的值就明確啦

// child.js
module.exports = cluster;

cluster.isWorker = true;
cluster.isMaster = false;

// master.js
module.exports = cluster;

cluster.isWorker = false;
cluster.isMaster = true;
複製代碼

那麼接下來的問題是: NODE_UNIQUE_ID從哪裏來?

1.2 NODE_UNIQUE_ID從哪裏來的？

在internal/cluster/master.js文件中搜索NODE_UNIQUE_ID ----> 上層爲createWorkerProcess函數 ----> 上層爲cluster.fork函數

master.js源碼中相關部分

const { fork } = require('child_process');

cluster.workers = {}

var ids = 0;

cluster.fork = function(env) {
  const id = ++ ids;
  const workerProcess = createWorkerProcess(id, env);
  const worker = new Worker({
    id: id,
    process: workerProcess
  });
  cluster.workers[worker.id] = worker;
  return worker
}

function createWorkerProcess(id, env) {
  const workerEnv = { ...process.env, ...env, NODE_UNIQUE_ID: `${id}` };
  
  return fork(args, {
    env: workerEnv
  })
}
複製代碼

結論: 變量NODE_UNIQUE_ID是在主進程fork子進程時傳遞進去的參數，所以採用cluster.fork建立的子進程是必定包含NODE_UNIQUE_ID的，而直接使用child_process.fork的子進程是沒有NODE_UNIQUE_ID的

而且, NODE_UNIQUE_ID將做爲主進程中存儲活躍的工做進程對象的鍵值

2. 主進程中是否存在TCP服務器, 若是有, 何時建立的？

繼續描述一下這個問題的由來:

const cluster = require('cluster');
const http = require('http');

if (cluster.isMaster) {
  // 根據cpu核心數出fork相同數量的子進程
} else {
  // 用http模塊建立server監聽某一個端口
}
複製代碼

並無在cluster.isMaster條件語句中建立服務器, 也沒有提供服務器相關的路徑，接口。而主進程又須要承擔代理服務器的 職責，那麼主進程中是否存在TCP服務器?

咱們來猜猜看可能的步驟

• 子進程會執行http.createServer

• http模塊會調用net模塊, 由於http.Server繼承net.Server

• 同時偵聽端口, 建立net.Server實例, 建立的實例調用listen(port), 等待連接

這時若是主進程要建立服務器就須要把建立服務器相關信息給主進程, 繼續猜想

• 假設主進程已經拿到了服務器相關的信息, 主進程本身來建立

• 後面的fork子進程就不用本身建立了，而是從主進程中get到相關數據

既然要在主進程須要獲得完整的建立服務器相關信息, 那麼極可能在net模塊listen相關方法中進行處理

2.1 在源碼中找答案

github net 模塊源碼

Server.prototype.listen找找看，何時把服務器相關信息傳遞給主進程了？

Server.prototype.listen = function(...args) {
  // 無視其餘的判斷邏輯, 直達它的心裏!
  if (成功) {
    listenInCluster()
    return this
  } else {
    // 無視
  }
}
複製代碼

總的來講就是: 在Server.prototype.listen函數中，在成功進入條件語句後全部的狀況都執行了listenInCluster函數後返回

接下來看listenInCluster函數

function listenInCluster(server, 建立服務器須要的數據) {
  if (cluster === undefined) cluster = require('cluster')
  // 判斷是不是主進程
  if (cluster.isMaster) {
    server._listen2(建立服務器須要的數據)
    return
  }
  // 建立服務器須要的數據
  const serverQuery = {
      address: address,
      port: port,
      addressType: addressType,
      fd: fd,
      flags,
    };
  // 只剩下子進程
  cluster._getServer(server, 建立服務器須要的數據, listenOnMasterHandle);
 
  function listenOnMasterHandle(err, handle) {
    server._handle = handle
    server._listen2(建立服務器須要的數據)
  }
}
複製代碼

按照前面的推斷: 子進程會給主進程發送建立server須要的數據, 主進程去建立

因此接下來去看cluster模塊的child._getServer函數

cluster._getServer = function(obj, options, cb) {
  // 組裝發送的數據
  const message = {
    act: 'queryServer',
    ...options,
  }
  // 發送數據
  send(message, (reply, handle) => {
  
  })
}
複製代碼

那麼接下來主進程就應該對queryServer做出想要的處理

具體能夠看cluster/master.js

const RoundRobinHandle = require('internal/cluster/round_robin_handle');

const handles = new Map()

function onmessage(message, handle) {
  if (message.act === 'queryServer') {
    queryServer(worker, message)
  }
}
queryServer(worker, message) {
  const key = `${message.address}:${message.port}:${message.addressType}:` +
              `${message.fd}:${message.index}`;
  const constructor = RoundRobinHandle
  
  let handle = new constructor(建立服務器相關信息)
  
  handles.set(key, handle);
}
複製代碼

終於要到終點了:

在internal/cluster/round_robin_handle.js中

function RoundRobinHandle(建立服務器相關信息) {
  this.server = net.createServer()
  this.server.listen(.....)
}
複製代碼

2.2 主進程在cluster模式下如何建立服務器的結論

主進程fork子進程, 子進程中有顯式建立服務器的操做，但實際上在cluster模式下, 子進程是把建立服務器所須要的數據發送給主進程, 主進程來隱式建立TCP服務器

流程圖

3. 爲何多個子進程能夠監聽同一個端口？

這個問題能夠轉換爲： 子進程中有沒有也建立一個服務器，同時偵聽某個端口呢？

其實，上面的源碼分析中能夠得出結論：子進程中確實建立了net.Server對象，但是它沒有像主進程那樣在libuv層構建socket句柄，子進程的net.Server對象使用的是一個假句柄來'欺騙'使用者端口已偵聽

3.1 首先要明確默認的調度策略: round-robin

這部分能夠參考文章Node.js V0.12 新特性之 Cluster 輪轉法負載均衡

主要就是說：Node.js v0.12引入了round-robin方式, 用輪轉法來分配請求, 每一個子進程的獲取的時間的機會都是均等的(windows除外)

源碼在internal/cluster/master.js中

var schedulingPolicy = {
  'none': SCHED_NONE,
  'rr': SCHED_RR
}[process.env.NODE_CLUSTER_SCHED_POLICY];

if (schedulingPolicy === undefined) {
  // FIXME Round-robin doesn't perform well on Windows right now due to the // way IOCP is wired up. schedulingPolicy = (process.platform === 'win32') ? SCHED_NONE : SCHED_RR; } cluster.schedulingPolicy = schedulingPolicy; 複製代碼

3.2 證實子進程拿到的是假句柄

上面說明了：默認的調度策略是round-robin, 那麼子進程將建立服務器的數據發送給主進程, 當主進程發送建立服務器成功的消息後，子進程會執行回調函數

源碼在internal/cluster/child.js _getServer中

cluster._getServer = function(obj, options, cb) {
  const indexesKey = [address,
                      options.port,
                      options.addressType,
                      options.fd ].join(':');
  send(message, (reply, handle) => {
    if (typeof obj._setServerData === 'function')
      obj._setServerData(reply.data);
      
    // 這裏能夠反推出主進程返回的handle爲null
    if (handle)
      shared(reply, handle, indexesKey, cb);  // Shared listen socket.
    else
      rr(reply, indexesKey, cb);              // Round-robin.
  });
}
複製代碼

rr函數, 注意這裏的回調函數其實就是net模塊中的listenOnMasterHandle方法

function rr(message, indexesKey, cb) {
  const key = message.key
  const handle = { close, listen, ref: noop, unref: noop };
  handles.set(key, handle)
  // 將假句柄傳遞給上層的net.Server
  cb(0, handle)
}
複製代碼

因此結論是這樣：子進程壓根沒有建立底層的服務端socket作偵聽，因此在子進程建立的HTTP服務器偵聽的端口根本不會出現端口複用的狀況

3.3 子進程沒有建立底層socket, 如何接收請求和發送響應呢？

顯而易見：主進程的服務器中會建立RoundRobinHandle決定分發請求給哪個子進程，篩選出子進程後發送newconn消息給對應的子進程

4. 請求分發策略 RoundRobin

源碼見internal/cluster/round_robin_handle

module.exports = RoundRobinHandle

function RoundRobinHandle(建立服務器須要的參數) {
  // 存儲空閒的子進程
  this.free = []
  // 存放待處理的用戶請求
  this.handles = []
}
// 負責篩選出處理請求的子進程
RoundRobinHandle.prototype.distribute = function(err, handle) {
  this.handles.push(handle)
  const worker = this.free.shift()
  
  if (worker) {
    this.handoff(worker)
  }
}
// 獲取請求，並經過IPC發送句柄handle和newconn消息，等待子進程返回
RoundRobinHandle.prototype.handoff = function(worker) {
  const handle = this.handles.shift()
  
  if (handle === undefined) {
    this.free.push(worker)
    return
  }
  
  const message = { act: 'newconn', key: this.key }
  
  sendHelper(worker.process, message, handle, reply => {
    if (reply.accepted)
      handle.close();
    // 某個子進程辦事不力，給下一個子進程再試試  
    else
      this.distribute(0, handle)  
    this.handoff(worker)  
  })
}
複製代碼

參考

Nodejs cluster模塊深刻探究