Node.js - 阿里Egg的多进程模型和进程间通讯

摘要：第二种是主进程创建监听后发送给感兴趣的工作进程，由工作进程负责直接接收连接。继续看，可以看到它捕获了事件，并在回调函数里面关闭连接，关闭本身进程，断开与的通道。参考与引用多进程模型和进程间通讯源码解析之

最近用Egg作为底层框架开发项目，好奇其多进程模型的管理实现，于是学习了解了一些东西，顺便记录下来。文章如有错误， 请轻喷

伴随科技的发展， 现在的服务器基本上都是多核cpu的了。然而，Node是一个单进程单线程语言（对于开发者来说是单线程，实际上不是）。我们都知道，cpu的调度单位是线程，而基于Node的特性，那么我们每次只能利用一个cpu。这样不仅仅利用率极低，而且容错更是不能接受（出错时会崩溃整个程序）。所以，Node有了cluster来协助我们充分利用服务器的资源。

cluster工作原理
关于cluster的工作原理推荐大家看这篇文章,这里简单总结一下：

子进程的端口监听会被hack掉，而是统一由master的内部TCP监听，所以不会出现多个子进程监听同一端口而报错的现象。

请求统一经过master的内部TCP，TCP的请求处理逻辑中，会挑选一个worker进程向其发送一个newconn内部消息，随消息发送客户端句柄。（这里的挑选有两种方式，第一种是除Windows外所有平台的默认方法循环法，即由主进程负责监听端口，接收新连接后再将连接循环分发给工作进程。在分发中使用了一些内置技巧防止工作进程任务过载。第二种是主进程创建监听socket后发送给感兴趣的工作进程，由工作进程负责直接接收连接。）

worker进程收到句柄后，创建客户端实例(net.socket)执行具体的业务逻辑，然后返回。

如图：

图引用出处

先看一下Egg官方文档的进程模型

                +--------+          +-------+
                | Master |<-------->| Agent |
                +--------+          +-------+
                ^   ^    ^
               /    |     
             /      |       
           /        |         
         v          v          v
+----------+   +----------+   +----------+
| Worker 1 |   | Worker 2 |   | Worker 3 |
+----------+   +----------+   +----------+

大致上就是利用Master作为主线程，启动Agent作为秘书进程协助Worker处理一些公共事务（日志之类），启动Worker进程执行真正的业务代码。

首先从Master入手，这里暂时认为Master是最顶级的进程（事实上还有一个parent进程，待会再说）。

/**
 * start egg app
 * @method Egg#startCluster
 * @param {Object} options {@link Master}
 * @param {Function} callback start success callback
 */
exports.startCluster = function(options, callback) {
  new Master(options).ready(callback);
};

先从Master的构造函数看起

constructor(options) {
  super();
  // 初始化参数
  this.options = parseOptions(options);
  // worker进程的管理类 详情见 Manager及Messenger篇
  this.workerManager = new Manager();
  // messenger类， 详情见 Manager及Messenger篇
  this.messenger = new Messenger(this);
  // 设置一个ready事件 详情见get-ready npm包
  ready.mixin(this);
  // 是否为生产环境
  this.isProduction = isProduction();
  this.agentWorkerIndex = 0;
  // 是否关闭
  this.closed = false;
  ...

  接下来看的是ready的回调函数及注册的各类事件：
  this.ready(() => {
    // 将开始状态设置为true
    this.isStarted = true;
    const stickyMsg = this.options.sticky ? " with STICKY MODE!" : "";
    this.logger.info("[master] %s started on %s (%sms)%s",
    frameworkPkg.name, this[APP_ADDRESS], Date.now() - startTime, stickyMsg);

    // 发送egg-ready至各个进程并触发相关事件
    const action = "egg-ready";
    this.messenger.send({ action, to: "parent", data: { port: this[REALPORT], address: this[APP_ADDRESS] } });
    this.messenger.send({ action, to: "app", data: this.options });
    this.messenger.send({ action, to: "agent", data: this.options });
    // start check agent and worker status
    this.workerManager.startCheck();
    });
    // 注册各类事件
    this.on("agent-exit", this.onAgentExit.bind(this));
    this.on("agent-start", this.onAgentStart.bind(this));
    ...
    // 检查端口并 Fork一个Agent
    detectPort((err, port) => {
      ... 
      this.forkAgentWorker();
    }
  });
}

综上， 可以看到Master的构造函数主要是初始化和注册各类相应的事件， 最后运行的是forkAgentWorker函数， 该函数的关键代码可以看到：

const agentWorkerFile = path.join(__dirname, "agent_worker.js");
// 通过child_process执行一个Agent
const agentWorker = childprocess.fork(agentWorkerFile, args, opt);

继续到agent_worker.js上面看，agent_worker实例化一个agent对象，agent_worker.js有一句关键代码：

agent.ready(() => {
  agent.removeListener("error", startErrorHandler); // 清除错误监听的事件
  process.send({ action: "agent-start", to: "master" }); // 向master发送一个agent-start的动作
});

可以看到， agent_worker.js中的代码向master发出了一个信息， 动作为agent-start， 再回到Master中， 可以看到其注册了两个事件， 分别为once的forkAppWorkers和 on的onAgentStart

this.on("agent-start", this.onAgentStart.bind(this));
this.once("agent-start", this.forkAppWorkers.bind(this));

先看onAgentStart函数， 这个函数相对简单， 就是一些信息的传递：

onAgentStart() {
    this.agentWorker.status = "started";

    // Send egg-ready when agent is started after launched
    if (this.isAllAppWorkerStarted) {
      this.messenger.send({ action: "egg-ready", to: "agent", data: this.options });
    }

    this.messenger.send({ action: "egg-pids", to: "app", data: [ this.agentWorker.pid ] });
    // should send current worker pids when agent restart
    if (this.isStarted) {
      this.messenger.send({ action: "egg-pids", to: "agent", data: this.workerManager.getListeningWorkerIds() });
    }

    this.messenger.send({ action: "agent-start", to: "app" });
    this.logger.info("[master] agent_worker#%s:%s started (%sms)",
      this.agentWorker.id, this.agentWorker.pid, Date.now() - this.agentStartTime);
  }

然后会执行forkAppWorkers函数，该函数主要是借助cfork包fork对应的工作进程， 并注册一系列相关的监听事件,

...
cfork({
  exec: this.getAppWorkerFile(),
  args,
  silent: false,
  count: this.options.workers,
  // don"t refork in local env
  refork: this.isProduction,
});
...
// 触发app-start事件
cluster.on("listening", (worker, address) => {
  this.messenger.send({
    action: "app-start",
    data: { workerPid: worker.process.pid, address },
    to: "master",
    from: "app",
  });
});

可以看到forkAppWorkers函数在监听Listening事件时，会触发master上的app-start事件。

this.on("app-start", this.onAppStart.bind(this));

...
// master ready回调触发
if (this.options.sticky) {
  this.startMasterSocketServer(err => {
    if (err) return this.ready(err);
      this.ready(true);
  });
} else {
  this.ready(true);
}

// ready回调 发送egg-ready状态到各个进程
const action = "egg-ready";
this.messenger.send({ action, to: "parent", data: { port: this[REALPORT], address: this[APP_ADDRESS] } });
this.messenger.send({ action, to: "app", data: this.options });
this.messenger.send({ action, to: "agent", data: this.options });

// start check agent and worker status
if (this.isProduction) {
  this.workerManager.startCheck();
}

总结下：

Master.constructor: 先执行Master的构造函数， 里面有个detect函数被执行

Detect: Detect => forkAgentWorker()

forkAgentWorker: 获取Agent进程， 向master触发agent-start事件

执行onAgentStart函数， 执行forkAppWorker函数（once）

onAgentStart => 发送各类信息, forkAppWorker => 向master触发 app-start事件

App-start事件 触发 onAppStart()方法

onAppStart => 设置ready(true) => 执行ready的回调函数

Ready() = > 发送egg-ready到各个进程并触发相关事件, 执行startCheck()函数

+---------+           +---------+          +---------+
|  Master |           |  Agent  |          |  Worker |
+---------+           +----+----+          +----+----+
     |      fork agent     |                    |
     +-------------------->|                    |
     |      agent ready    |                    |
     |<--------------------+                    |
     |                     |     fork worker    |
     +----------------------------------------->|
     |     worker ready    |                    |
     |<-----------------------------------------+
     |      Egg ready      |                    |
     +-------------------->|                    |
     |      Egg ready      |                    |
     +----------------------------------------->|

根据官方文档，进程守护主要是依赖于graceful和egg-cluster这两个库。

未捕获异常

关闭异常 Worker 进程所有的 TCP Server（将已有的连接快速断开，且不再接收新的连接），断开和 Master 的 IPC 通道，不再接受新的用户请求。

Master 立刻 fork 一个新的 Worker 进程，保证在线的『工人』总数不变。

异常 Worker 等待一段时间，处理完已经接受的请求后退出。

+---------+                 +---------+
|  Worker |                 |  Master |
+---------+                 +----+----+
     | uncaughtException         |
     +------------+              |
     |            |              |                   +---------+
     | <----------+              |                   |  Worker |
     |                           |                   +----+----+
     |        disconnect         |   fork a new worker    |
     +-------------------------> + ---------------------> |
     |         wait...           |                        |
     |          exit             |                        |
     +-------------------------> |                        |
     |                           |                        |
    die                          |                        |
                                 |                        |
                                 |                        |

由执行的app文件可知， app实际上是继承于Application类, 该类下面调用了graceful()。

onServer(server) {
    ......
    graceful({
      server: [ server ],
      error: (err, throwErrorCount) => {
        ......
      },
    });
    ......
  }

继续看graceful， 可以看到它捕获了process.on("uncaughtException")事件， 并在回调函数里面关闭TCP连接， 关闭本身进程， 断开与master的IPC通道。

process.on("uncaughtException", function (err) {
    ......
    // 对http连接设置 Connection: close响应头
    servers.forEach(function (server) {
      if (server instanceof http.Server) {
        server.on("request", function (req, res) {
          // Let http server set `Connection: close` header, and close the current request socket.
          req.shouldKeepAlive = false;
          res.shouldKeepAlive = false;
          if (!res._header) {
            res.setHeader("Connection", "close");
          }
        });
      }
    });

    // 设置一个定时函数关闭子进程， 并退出本身进程
    // make sure we close down within `killTimeout` seconds
    var killtimer = setTimeout(function () {
      console.error("[%s] [graceful:worker:%s] kill timeout, exit now.", Date(), process.pid);
      if (process.env.NODE_ENV !== "test") {
        // kill children by SIGKILL before exit
        killChildren(function() {
          // 退出本身进程
          process.exit(1);
        });
      }
    }, killTimeout);

    // But don"t keep the process open just for that!
    // If there is no more io waitting, just let process exit normally.
    if (typeof killtimer.unref === "function") {
      // only worked on node 0.10+
      killtimer.unref();
    }

    var worker = options.worker || cluster.worker;

    // cluster mode
    if (worker) {
      try {
        // 关闭TCP连接
        for (var i = 0; i < servers.length; i++) {
          var server = servers[i];
          server.close();
        }
      } catch (er1) {
        ......
      }

      try {
        // 关闭ICP通道
        worker.disconnect();
      } catch (er2) {
        ......
      }
    }
  });

ok, 关闭了IPC通道后， 我们继续看cfork文件， 即上面提到的fork worker的包， 里面监听了子进程的disconnect事件， 他会根据条件判断是否重新fork一个新的子进程

cluster.on("disconnect", function (worker) {
    ......
    // 存起该pid
    disconnects[worker.process.pid] = utility.logDate();
    if (allow()) {
      // fork一个新的子进程
      newWorker = forkWorker(worker._clusterSettings);
      newWorker._clusterSettings = worker._clusterSettings;
    } else {
      ......
    }
  });

一般来说， 这个时候会继续等待一会然后就执行了上面说到的定时函数了， 即退出进程。

OOM、系统异常
关于这种系统异常， 有时候在子进程中是不能捕获到的， 我们只能在master中进行处理， 也就是cfork包。

cluster.on("exit", function (worker, code, signal) {
    // 是程序异常的话， 会通过上面提到的uncatughException重新fork一个子进程， 所以这里就不需要了
    var isExpected = !!disconnects[worker.process.pid];
    if (isExpected) {
      delete disconnects[worker.process.pid];
      // worker disconnect first, exit expected
      return;
    }
    // 是master杀死的子进程， 无需fork
    if (worker.disableRefork) {
      // worker is killed by master
      return;
    }

    if (allow()) {
      newWorker = forkWorker(worker._clusterSettings);
      newWorker._clusterSettings = worker._clusterSettings;
    } else {
      ......
    }
    cluster.emit("unexpectedExit", worker, code, signal);
  });

上面一直提到各种进程间通信，细心的你可能已经发现 cluster 的 IPC 通道只存在于 Master 和 Worker/Agent 之间，Worker 与 Agent 进程互相间是没有的。那么 Worker 之间想通讯该怎么办呢？是的，通过 Master 来转发。

广播消息： agent => all workers
                  +--------+          +-------+
                  | Master |<---------| Agent |
                  +--------+          +-------+
                 /    |     
                /     |      
               /      |       
              /       |        
             v        v         v
  +----------+   +----------+   +----------+
  | Worker 1 |   | Worker 2 |   | Worker 3 |
  +----------+   +----------+   +----------+

指定接收方： one worker => another worker
                  +--------+          +-------+
                  | Master |----------| Agent |
                  +--------+          +-------+
                 ^    |
     send to    /     |
    worker 2   /      |
              /       |
             /        v
  +----------+   +----------+   +----------+
  | Worker 1 |   | Worker 2 |   | Worker 3 |
  +----------+   +----------+   +----------+

在master中， 可以看到当agent和app被fork时， 会监听他们的信息， 同时将信息转化成一个对象:

agentWorker.on("message", msg => {
  if (typeof msg === "string") msg = { action: msg, data: msg };
  msg.from = "agent";
  this.messenger.send(msg);
});

worker.on("message", msg => {
  if (typeof msg === "string") msg = { action: msg, data: msg };
  msg.from = "app";
  this.messenger.send(msg);
});

可以看到最后调用的是messenger.send, 而messengeer.send就是根据from和to来决定将信息发送到哪里

send(data) {
    if (!data.from) {
      data.from = "master";
    }
    ......

    // app -> master
    // agent -> master
    if (data.to === "master") {
      debug("%s -> master, data: %j", data.from, data);
      // app/agent to master
      this.sendToMaster(data);
      return;
    }

    // master -> parent
    // app -> parent
    // agent -> parent
    if (data.to === "parent") {
      debug("%s -> parent, data: %j", data.from, data);
      this.sendToParent(data);
      return;
    }

    // parent -> master -> app
    // agent -> master -> app
    if (data.to === "app") {
      debug("%s -> %s, data: %j", data.from, data.to, data);
      this.sendToAppWorker(data);
      return;
    }

    // parent -> master -> agent
    // app -> master -> agent，可能不指定 to
    if (data.to === "agent") {
      debug("%s -> %s, data: %j", data.from, data.to, data);
      this.sendToAgentWorker(data);
      return;
    }
  }

master则是直接根据action信息emit对应的注册事件

sendToMaster(data) {
  this.master.emit(data.action, data.data);
}

而agent和worker则是通过一个sendmessage包， 实际上就是调用下面类似的方法

 // 将信息传给子进程
 agent.send(data)
 worker.send(data)

最后， 在agent和app都继承的基础类EggApplication上， 调用了Messenger类， 该类内部的构造函数如下：

constructor() {
    super();
    ......
    this._onMessage = this._onMessage.bind(this);
    process.on("message", this._onMessage);
  }

_onMessage(message) {
    if (message && is.string(message.action)) {
      // 和master一样根据action信息emit对应的注册事件  
      this.emit(message.action, message.data);
    }
  }

总结一下：
思路就是利用事件机制和IPC通道来达到各个进程之间的通信。

学习过程中有遇到一个timeout.unref()的函数， 关于该函数推荐大家参考这个问题的6楼回答

从前端思维转到后端思维其实还是很吃力的，加上Egg的进程管理实现确实非常厉害， 所以花了很多时间在各种api和思路思考上。

多进程模型和进程间通讯
Egg 源码解析之 egg-cluster