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

Cheng_Gang 发布于2019-08-23 16:58 / 954人阅读

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

前言

最近用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	1	进程管理，进程间消息转发	非常高	否
Agent	1	后台运行工作（长连接客户端）	高	少量
Worker	一般为cpu核数	执行业务代码	一般	是

大致上就是利用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);
  });

进程间通信（IPC）

</>复制代码 
上面一直提到各种进程间通信，细心的你可能已经发现 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

GPU云服务器云服务器进程间通讯动态库进程间通讯linux 进程间通信 IPC进程间通信

文章版权归作者所有，未经允许请勿转载,若此文章存在违规行为，您可以联系管理员删除。

转载请注明本文地址：https://www.ucloud.cn/yun/103585.html

前端最实用书签(持续更新)

摘要：前言一直混迹社区突然发现自己收藏了不少好文但是管理起来有点混乱所以将前端主流技术做了一个书签整理不求最多最全但求最实用。前言一直混迹社区,突然发现自己收藏了不少好文但是管理起来有点混乱; 所以将前端主流技术做了一个书签整理,不求最多最全,但求最实用。书签源码书签导入浏览器效果截图showImg(https://segmentfault.com/img/bVbg41b?w=107...

sshe 2019-08-23 12:00 评论0 收藏0
实时联网游戏后台服务技术选型与挑战（网络接入篇）

摘要：概述本系列文章将从开发者角度梳理开发实时联网游戏后台服务过程中可能面临的挑战，并针对性地提供相应解决思路，期望帮助开发者依据自身游戏特点做出合理的技术选型。多路复用避免了读写阻塞，减少了上下文切换，提升了利用率和系统吞吐率。概述：本系列文章将从开发者角度梳理开发实时联网游戏后台服务过程中可能面临的挑战，并针对性地提供相应解决思路，期望帮助开发者依据自身游戏特点做出合理的技术选型。关...

zhisheng 2019-08-16 12:32 评论0 收藏0
PHP进程通信

摘要：一进程间通信理解间进程通信机制，先了解下进程间有哪些通讯机制历史发展按照历史来源主要有两大块的管道，，信号的消息队列，共享内存，信号灯。信号量主要作为进程间，以及进程内部线程之间的通讯手段。主要依赖，兼容扩展实现方式的进程间通信之消息队列。 PHP间进程如何通信，PHP相关的服务的IPC是实现方式，IPC的思想如何用到项目中。一、linux进程间通信理解php间进程通信机制，先了解...

haobowd 2019-06-27 16:53 评论0 收藏0
系列３｜走进Node.js之多进程模型

摘要：例如，在方法中，如果需要主从进程之间建立管道，则通过环境变量来告知从进程应该绑定的相关的文件描述符，这个特殊的环境变量后面会被再次涉及到。文：正龙（沪江网校Web前端工程师）本文原创，转载请注明作者及出处之前的文章走进Node.js之HTTP实现分析中，大家已经了解 Node.js 是如何处理 HTTP 请求的，在整个处理过程，它仅仅用到单进程模型。那么如何让 Web 应用扩展到...

snowell 2019-08-26 11:30 评论0 收藏0
从单租户IaaS到多租户PaaS——金融级别大数据平台MaxCompute的多租户隔离实践

摘要：摘要在年云栖大会北京峰会的大数据专场中，来自阿里云的高级技术专家李雪峰带来了主题为金融级别大数据平台的多租户隔离实践的演讲。三是运行隔离机制。针对这一问题，提供了多层隔离嵌套方案以便规避这种潜在的安全风险。摘要：在2017年云栖大会•北京峰会的大数据专场中，来自阿里云的高级技术专家李雪峰带来了主题为《金融级别大数据平台的多租户隔离实践》的演讲。在分享中，李雪峰首先介绍了基于传统Iaa...

beanlam 2019-06-24 17:58 评论0 收藏0