node集群（cluster）

使用例子

為了讓node應用能夠在多核伺服器中提高性能，node提供cluster API，用於創建多個工作進程，然後由這些工作進程並行處理請求。

// master.js
const cluster = require('cluster');
const cpusLen = require('os').cpus().length;
const path = require('path');

console.log(`主進程：${process.pid}`);
cluster.setupMaster({
  exec: path.resolve(__dirname, './work.js'),
});

for (let i = 0; i < cpusLen; i++) {
  cluster.fork();
}

// work.js
const http = require('http');

console.log(`工作進程：${process.pid}`);
http.createServer((req, res) => {
  res.end('hello');
}).listen(8080);

上面例子中，使用cluster創建多個工作進程，這些工作進程能夠共用8080埠，我們請求localhost:8080，請求任務會交給其中一個工作進程進行處理，該工作進程處理完成後，自行響應請求。

埠佔用問題

這裡有個問題，前面例子中，出現多個進程監聽相同的埠，為什麼程式沒有報埠佔用問題，由於socket套接字監聽埠會有一個文件描述符，而每個進程的文件描述符都不相同，無法讓多個進程都監聽同一個埠，如下：

// master.js
const fork = require('child_process').fork;
const cpusLen = require('os').cpus().length;
const path = require('path');

console.log(`主進程：${process.pid}`);
for (let i = 0; i < cpusLen; i++) {
  fork(path.resolve(__dirname, './work.js'));
}

// work.js
const http = require('http');

console.log(`工作進程：${process.pid}`);
http.createServer((req, res) => {
  res.end('hello');
}).listen(8080);

當運行master.js文件的時候，會報埠被佔用的問題（Error: listen EADDRINUSE: address already in use :::8080）。

我們修改下，只使用主進程監聽埠，主進程將請求套接字發放給工作進程，由工作進程來進行業務處理。

// master.js
const fork = require('child_process').fork;
const cpusLen = require('os').cpus().length;
const path = require('path');
const net = require('net');
const server = net.createServer();

console.log(`主進程：${process.pid}`);
const works = [];
let current = 0
for (let i = 0; i < cpusLen; i++) {
  works.push(fork(path.resolve(__dirname, './work.js')));
}

server.listen(8080, () => {
  if (current > works.length - 1) current = 0
  works[current++].send('server', server);
  server.close();
});

// work.js
const http = require('http');
const server = http.createServer((req, res) => {
  res.end('hello');
});

console.log(`工作進程：${process.pid}`);
process.on('message', (type, tcp) => {
  if (type === 'server') {
    tcp.on('connection', socket => {
      server.emit('connection', socket)
    });
  }
})

實際上，cluster新建的工作進程並沒有真正去監聽埠，在工作進程中的net server listen函數會被hack，工作進程調用listen，不會有任何效果。監聽埠工作交給了主進程，該埠對應的工作進程會被綁定到主進程中，當請求進來的時候，主進程會將請求的套接字下發給相應的工作進程，工作進程再對請求進行處理。

接下來我們看看cluster API中的實現，看下cluster內部是如何做到下面兩個功能：

  • 主進程：對傳入的埠進行監聽
  • 工作進程：
    • 主進程註冊當前工作進程，如果主進程是第一次監聽此埠，就新建一個TCP伺服器，並將當前工作進程和TCP伺服器綁定。
    • hack掉工作進程中的listen函數，讓該進程不能監聽埠

源碼解讀

本文使用的是[email protected]。

// lib/cluster.js
'use strict';

const childOrPrimary = 'NODE_UNIQUE_ID' in process.env ? 'child' : 'primary';
module.exports = require(`internal/cluster/${childOrPrimary}`);

這個是cluster API入口，在引用cluster的時候，程式首先會判斷環境變數中是否存在NODE_UNIQUE_ID變數，來確定當前程式是在主進程運行還是工作進程中運行。NODE_UNIQUE_ID實際上就是一個自增的數字，是工作進程的ID，後面會在創建工作進程相關程式碼中看到，這裡就不多做解釋了。

通過前面程式碼我們知道，如果在主進程中引用cluster，程式導出的是internal/cluster/primary.js這文件，因此我們先看看這個文件內部的一些實現。

// internal/cluster/primary.js
// ...
const EventEmitter = require('events');
const cluster = new EventEmitter();
// 下面這三個參數會在node內部功能實現的時候用到，之後我們看net源碼的時候會用到這些參數
cluster.isWorker = false; // 是否是工作進程
cluster.isMaster = true; // 是否是主進程
cluster.isPrimary = true; // 是否是主進程

module.exports = cluster;

cluster.setupPrimary = function(options) {
  const settings = {
    args: ArrayPrototypeSlice(process.argv, 2),
    exec: process.argv[1],
    execArgv: process.execArgv,
    silent: false,
    ...cluster.settings,
    ...options
  };

  cluster.settings = settings;
  // ...
}

cluster.setupMaster = cluster.setupPrimary;

cluster.fork = function(env) {
  cluster.setupPrimary();
  const id = ++ids;
  const workerProcess = createWorkerProcess(id, env);
}

const { fork } = require('child_process');
function createWorkerProcess(id, env) {
  // 這裡的NODE_UNIQUE_ID就是入口文件用來分辨當前進程類型用的
  const workerEnv = { ...process.env, ...env, NODE_UNIQUE_ID: `${id}` };
  // ...
  return fork(cluster.settings.exec, cluster.settings.args, {
    env: workerEnv,
    // ...
  });
}

cluster.fork用來新建一個工作進程，其內部使用child_process中的fork函數，來創建一個進程，創建的新進程默認會運行命令行中執行的入口文件（process.argv[1]），當然我們也可以執行luster.setupPrimary或者cluster.setupMaster並傳入exec參數來修改工作進程執行的文件。

我們再來簡單看下工作進程引用的internal/cluster/child.js文件：

// internal/cluster/child.js
const EventEmitter = require('events');
const cluster = new EventEmitter();

module.exports = cluster;
// 這裡定義的就是一個工作進程，後續會用到這裡的參數
cluster.isWorker = true;
cluster.isMaster = false;
cluster.isPrimary = false;

cluster._getServer = function(obj, options, cb) {
  // ...
};
// ...

這裡我們主要記住工作進程中的cluster有個_getServer函數，後續流程走到這個函數的時候，會詳細看裡面的程式碼。

接下來進入正題，看下net server listen函數：

// lib/net.js
Server.prototype.listen = function(...args) {
  // ...
  if (typeof options.port === 'number' || typeof options.port === 'string') {
    // 如果是向最開始那種直接調用listen時直接傳入一個埠，就會直接進入else，我們也主要看else中的邏輯
    if (options.host) {
      // ...
    } else  {
      // listen(8080, () => {...})調用方式，將運行這條分支
      listenInCluster(this, null, options.port | 0, 4, backlog, undefined, options.exclusive);
    }
    return this;
  }
  // ...
}

function listenInCluster(server, address, port, addressType, backlog, fd, exclusive, flags) {
  // ...
  // 這裡就用到cluster初始時寫入的isPrimary參數，當前如果在主進程isPrimary就為true，反之為false。主進程會直接去執行server._listen2函數，工作進程之後也會執行這個函數，等下一起看server._listen2內部的功能。
  if (cluster.isPrimary || exclusive) {
    server._listen2(address, port, addressType, backlog, fd, flags);
    return;
  }

  // 後面的程式碼只有在工作進程中才會執行
  const serverQuery = {
    address: address,
    port: port,
    addressType: addressType,
    fd: fd,
    flags,
  };

  // 這裡執行的是internal/cluster/child.js中的cluster._getServer，同時會傳入listenOnPrimaryHandle這個回調函數，這個回調函數會在主進程添加埠監聽，同時將工作進程綁定到對應的TCP服務後才會執行，裡面工作就是對net server listen等函數進行hack。
  cluster._getServer(server, serverQuery, listenOnPrimaryHandle);

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

// 等工作進程執行這個函數的時候再一起講
Server.prototype._listen2 = setupListenHandle;
function setupListenHandle(...) {
  // ...
}

從上面程式碼中可以得知，主進程和工作進程中執行net server listen都會進入到一個setupListenHandle函數中。不過區別是，主進程是直接執行該函數，而工作進程需要先執行cluster._getServer函數，讓主進程監聽工作進程埠，同時對listen函數進行hack處理，然後再執行setupListenHandle函數。接下來我們看下cluster._getServer函數的內部實現。

// lib/internal/cluster/child.js
cluster._getServer = function(obj, options, cb) {
  // ...
  // 這個是工作進程第一次發送內部消息的內容。
  // 注意這裡act值為queryServer
  const message = {
    act: 'queryServer',
    index,
    data: null,
    ...options
  };
  // ...
  // send函數內部使用IPC信道向工作進程發送內部消息。主進程在使用cluster.fork新建工作進程的時候，會讓工作進程監聽內部消息事件，下面會展示具體程式碼
  // send調用傳入的回調函數會被寫入到lib/internal/cluster/utils.js文件中的callbacks map中，等後面要用的時候，再提取出來。
  send(message, (reply, handle) => {
    if (typeof obj._setServerData === 'function')
      obj._setServerData(reply.data);

    if (handle)
      shared(reply, handle, indexesKey, index, cb);
    else
      // 這個函數內部會定義一個listen函數，用來hack net server listen函數
      rr(reply, indexesKey, index, cb);
  });
  // ...
}

function send(message, cb) {
  return sendHelper(process, message, null, cb);
}

// lib/internal/cluster/utils.js
// ...
const callbacks = new SafeMap();
let seq = 0;
function sendHelper(proc, message, handle, cb) {
  message = { cmd: 'NODE_CLUSTER', ...message, seq };

  if (typeof cb === 'function')
    // 這裡將傳入的回調函數記錄下來。
    // 注意這裡的key是遞增數字
    callbacks.set(seq, cb);

  seq += 1;
  // 利用IPC信道，給當前工作進程發送內部消息
  return proc.send(message, handle);
}
// ...

工作進程中cluster._getServer函數執行，將生成一個回調函數，將這個回調函數存放起來，並且會使用IPC信道，向當前工作進程發送內部消息。主進程執行cluster.fork生成工作進程的時候，會在工作進程中註冊internalMessage事件。接下來我們看下cluster.fork中與工作進程註冊內部消息事件的程式碼。

// internal/cluster/primary.js
cluster.fork = function(env) {
  // ...
  // internal函數執行會返回一個接收message對象的回調函數。
  // 可以先看下lib/internal/cluster/utils.js中的internal函數，了解內部的工作
  worker.process.on('internalMessage', internal(worker, onmessage));
  // ...
}

const methodMessageMapping = {
  close,
  exitedAfterDisconnect,
  listening,
  online,
  queryServer,
};

// 第一次觸發internalMessage執行的回調是這個函數。
// 此時message的act為queryServer
function onmessage(message, handle) {
  // internal內部在執行onmessage時會將這個函數執行上下文綁定到工作進程的work上
  const worker = this;

  // 工作進程傳入的
  const fn = methodMessageMapping[message.act];

  if (typeof fn === 'function')
    fn(worker, message);
}

function queryServer(worker, message) {
  // ...
}

// lib/internal/cluster/utils.js
// ...
const callbacks = new SafeMap();

function internal(worker, cb) {
  return function onInternalMessage(message, handle) {
    let fn = cb;

    // 工作進程第一次發送內部消息：ack為undefined，callback為undefined，直接執行internal調用傳入的onmessage函數，message函數只是用於解析消息的，實際會執行queryServer函數
    // 工作進程第二次發送內部消息：主進程queryServer函數執行會用工作進程發送內部消息，並向message中添加ack參數，讓message.ack=message.seq
    if (message.ack !== undefined) {
      const callback = callbacks.get(message.ack);

      if (callback !== undefined) {
        fn = callback;
        callbacks.delete(message.ack);
      }
    }

    ReflectApply(fn, worker, arguments);
  };
}

工作進程第一次發送內部消息時，由於傳入的message.ack（這裡注意分清act和ack）為undefind，因此沒辦法直接拿到cluster._getServer中調用send寫入的回調函數，因此只能先執行internal/cluster/primary.js中的queryServer函數。接下來看下queryServer函數內部邏輯。

// internal/cluster/primary.js
// hadles中存放的就是TCP伺服器。
// 主進程在代替工作進程監聽埠生成新的TCP伺服器前，
// 需要先判斷該伺服器是否有創建，如果有，就直接復用之前的伺服器，然後將工作進程綁定到相應的伺服器上；如果沒有，就新建一個TCP伺服器，然後將工作進程綁定到新建的伺服器上。
function queryServer(worker, message) {
  // 這裡key就是伺服器的唯一標識
  const key = `${message.address}:${message.port}:${message.addressType}:` +
              `${message.fd}:${message.index}`;
  // 從現存的伺服器中查看是否有當前需要的伺服器
  let handle = handles.get(key);
  // 如果沒有需要的伺服器，就新建一個
  if (handle === undefined) {
    // ...
    // RoundRobinHandle構建函數中，會新建一個TCP伺服器
    let constructor = RoundRobinHandle;
    handle = new constructor(key, address, message);
    // 將這個伺服器存放起來
    handles.set(key, handle);
  }

  if (!handle.data)
    handle.data = message.data;

  // 可以先看下下面關於RoundRobinHandle構建函數的程式碼，了解內部機制
  handle.add(worker, (errno, reply, handle) => {
    const { data } = handles.get(key);

    if (errno)
      handles.delete(key);

    // 這裡會向工作進程中發送第二次內部消息。
    // 這裡只傳了worker和message，沒有傳入handle和cb
    send(worker, {
      errno,
      key,
      ack: message.seq, // 注意這裡增加了ack屬性
      data,
      ...reply
    }, handle);
  });
}
function send(worker, message, handle, cb) {
  return sendHelper(worker.process, message, handle, cb);
}

// internal/cluster/round_robin_handle.js
function RoundRobinHandle(key, address, { port, fd, flags }) {
  // ...
  this.server = net.createServer(assert.fail);
  if (fd >= 0)
    this.server.listen({ fd });
  else if (port >= 0) {
    this.server.listen({
      port,
      host: address,
      ipv6Only: Boolean(flags & constants.UV_TCP_IPV6ONLY),
    });
  } else
    this.server.listen(address);

  // 當服務處於監聽狀態，就會執行這個回調。
  this.server.once('listening', () => {
    this.handle = this.server._handle;
    this.handle.onconnection = (err, handle) => this.distribute(err, handle);
    this.server._handle = null;
    // 注意：如果監聽成功，就會將server刪除
    this.server = null;
  });
}

RoundRobinHandle.prototype.add = function(worker, send) {
  const done = () => {
    if (this.handle.getsockname) {
      // ...
      send(null, { sockname: out }, null);
    } else {
      send(null, null, null);  // UNIX socket.
    }
    // ...
  };

  // 如果在add執行前server就已經處於listening狀態，this.server就會為null
  if (this.server === null)
    return done();
  // 如果add執行後，server才處於listening，就會走到這裡，始終都會執行add調用時傳入的回調
  this.server.once('listening', done);
}

在這一步，主進程替工作進程生成或者是獲取了一個可用的TCP伺服器，並將工作進程與相應的伺服器綁定在一起（方便後續請求任務分配）。當工作進程綁定完成以後，就向工作進程中發送了第二次內部消息，接下來我們再次進入lib/internal/cluster/utils.js看看內部流程：

// lib/internal/cluster/utils.js
const callbacks = new SafeMap();

function internal(worker, cb) {
  // 注意這裡handle為undefined
  return function onInternalMessage(message, handle) {
    let fn = cb;

    // 第二次工作進程內部消息執行的時候message.ack已經被賦值為message.seq
    // 因此這次能夠獲取到之前lib/cluster.child.js cluster._getServer函數執行是調用send寫入的回調函數
    if (message.ack !== undefined) {
      const callback = callbacks.get(message.ack);

      if (callback !== undefined) {
        fn = callback;
        callbacks.delete(message.ack);
      }
    }

    ReflectApply(fn, worker, arguments);
  };
}

工作進程第二次接受到內部消息時，cluster._getServer函數執行是調用send寫入的回調函數會被執行，接下來看下send寫入的回調函數內容：

// lib/internal/cluster/child.js
send(message, (reply, handle) => {
  // 此時handle為undefined，流程會直接運行rr函數
  if (handle)
    shared(reply, handle, indexesKey, index, cb); 
  else
    // 這裡的cb是lib/net.js在執行cluster._getServer時傳入listenOnPrimaryHandle函數，後面會介紹他的工作。
    rr(reply, indexesKey, index, cb);
});

function rr(message, indexesKey, index, cb) {
  let key = message.key;

  // 這裡定義的listen用於hack net server.listen，在工作進程中執行listen，工作進程並不會真正去監聽埠
  function listen(backlog) {
    return 0;
  }

  function close() {...}

  function getsockname(out) {...}

  const handle = { close, listen, ref: noop, unref: noop };
  handles.set(key, handle);
  // 執行傳入的listenOnPrimaryHandle函數
  cb(0, handle);
}

rr函數執行，會新建幾個與net server中同名的函數，並通過handle傳入listenOnPrimaryHandle函數。

// lib/net.js
function listenInCluster(...) {
  cluster._getServer(server, serverQuery, listenOnPrimaryHandle);

  // listenOnPrimaryHandle函數中將工作進程生成的server._handle對象替換成自定義的handle對象，後續server listen執行的就是server._handle中的listen函數，因此這裡就完成了對工作進程中的listen函數hack
  function listenOnPrimaryHandle(err, handle) {
    // ...
    // handle：{ listen: ..., close: ...., ... }
    server._handle = handle;
    server._listen2(address, port, addressType, backlog, fd, flags);
  }
}

下面看下server._listen2函數執行內容

Server.prototype._listen2 = setupListenHandle;

function setupListenHandle(address, port, addressType, backlog, fd, flags) {
  // 忽略，只要是從工作進程進來的，this._handle就是自己定義的對象內容
  if (this._handle) {
    debug('setupListenHandle: have a handle already');
  } else {
    // 主進程會進入這一層邏輯，會在這裡生成一個伺服器
    // ...
    rval = createServerHandle(address, port, addressType, fd, flags);
    // ...
    this._handle = rval;
  }
  const err = this._handle.listen(backlog || 511);
  // ...
}

至此，工作進程埠監聽相關的源碼就看完了，現在差不多了解到工作進程中執行net server listen時，工作進程並不會真正去監聽埠，埠監聽工作始終會交給主進程來完成。主進程在接到工作進程發來的埠監聽的時候，首先會判斷是否有相同的伺服器，如果有，就直接將工作進程綁定到對應的伺服器上，這樣就不會出現埠被佔用的問題；如果沒有對應的伺服器，就生成一個新的服務。主進程接受到請求的時候，就會將請求任務分配給工作進程，如何分配，就需要看具體使用的哪種負載均衡了。

Tags:

VirMach 便宜 VPS

QNews

QNews