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 |+———-+ +———-+ +———-+类型进程数量作用稳定性是否运行业务代码Master1进程管理，进程间消息转发非常高否Agent1后台运行工作（长连接客户端）高少量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执行一个Agentconst 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的onAgentStartthis.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 statusif (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