Node.js集群模块cluster之工作进程类Worker介绍

 2015年11月28日    902     声明


集群模块cluster可以创建进程集群,从而使Node程序有效的利用多核资源。创建进程集群使用cluster.fork()方法,该方法每调用一次会创建一个工作进程对象,每个工作进程对象是一个Worker类。

  1. Worker类中的事件
  2. Worker类中的属性
  3. Worker类中的方法


Worker对象包含了工作进程所有公开的信息和方法,可以通过主进程中的cluster.workers或工作进程中的 cluster.worker访问这个对象。

1. Worker类中的事件

1.1 事件:'disconnect'

和主进程的cluster.on('disconnect')事件一样,但仅发生在特定的状态改变的工作进程中。

cluster.fork().on('disconnect', function() {
  // Worker 断开了连接
});


1.2 事件:'error'

child_process.fork()'error'事件一样。在worker对象中同样可以使用process.on('error')事件进行处理。


1.3 事件:'exit'

  • code {Number} 如果是正常退出则为退出代码。
  • signal {String} 使得进程被终止的信号的名称(比如 'SIGHUP')。

某个工作进程实例在底层子进程被结束时触发此事件,与cluster.on('exit')事件一样,但发生在工作进程中。

var worker = cluster.fork();
worker.on('exit', function(code, signal) {
  if( signal ) {
    console.log("worker 进程被杀死,信号: "+signal);
  } else if( code !== 0 ) {
    console.log("worker退出了,错误码: "+code);
  } else {
    console.log("工作完成!");
  }
});


1.4 事件:'listening'

  • address {Object} 。

某个工作进程启动临听时触发此事件,与cluster.on('listening')事件一样,但发生在工作进程中。

cluster.fork().on('listening', function(address) {
  // Worker 监听中
});


1.5 事件:'message'

  • message {Object} 。

该事件和 child_process.fork() 所提供的一样。在主进程中应当使用这个事件,而在工作进程中可以使用这个事件也可以使用 process.on('message')

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

if (cluster.isMaster) {

  // 跟踪http请求数
  var numReqs = 0;
  setInterval(function() {
    console.log("numReqs =", numReqs);
  }, 1000);

  // 统计请求数
  function messageHandler(msg) {
    if (msg.cmd && msg.cmd == 'notifyRequest') {
      numReqs += 1;
    }
  }

  // 启动 workers 并监听信息中有多少个notifyRequest
  var numCPUs = require('os').cpus().length;
  for (var i = 0; i < numCPUs; i++) { 
  	cluster.fork(); 
  } 
  Object.keys(cluster.workers).forEach(function(id) { 
  	cluster.workers[id].on('message', messageHandler); 
  }); 
} else { 
  // Worker 进程中有一个 http 服务器 
  http.Server(function(req, res) { 
  	res.writeHead(200); 
  	res.end("hello world\n"); 
  	// 通知主进程请求数 
  	process.send({ cmd: 'notifyRequest' }); 
  }).listen(8000); 
}


1.6 事件:'online'

  • address {Object} 。

cluster.on('online')事件一样,但发生在工作进程中。

cluster.fork().on('online', function() {
  // Worker is online
});


2. Worker类中的属性

2.1 进程ID:worker.id

  • 返回值:{Number}

每个工作进程都会被分配一个唯一的标识,这个标识被存储在id属性中,这个属性同样是cluster.workers对象的key值。


2.2 进程对象:worker.process

  • 返回值:{ChildProcess object}

所有的工作进程本质上都是通过child_process.fork()创建的,该函数返回的对象被储存在process中。


2.3 进程对象:worker.suicide

  • 返回值:{Blooean}

调用 .kill().disconnect() 后设置为true,在这之前是 undefined

通过这个属性,可以区分出是正常退出还是意外退出,主进程可以根据这个值,来决定是否是重新派生成工作进程

cluster.on('exit', function(worker, code, signal) {
  if (worker.suicide === true) {
    console.log('这是个自杀的进程\' – 不要担心').
  }
});

// 杀死worker
worker.kill();


3. Worker类中的方法

3.1 关闭所有工作进程:worker.disconnect()

worker.disconnect()

在工作进程中,这个函数会关闭所有服务器,收到服务器的'close'事件,关闭 IPC 通道。

在主进程中,会发给工作进程一个内部消息,用于调用.disconnect()方法结束工作进程

调用这个方法后,.suicide会被设置。 注意,服务器关闭后,不再接受新的连接,但可以接受新的监听。已经存在的连接允许正常退出。当连接为空得时候,工作进程的 IPC 通道会优雅的退出。

以上仅适用于服务器的连接,客户端的连接由工作进程关闭。 由于长连接可能会阻塞进程关闭连接,有一个较好的办法是发消息给应用,这样应用会想办法关闭它们。也可以使用超时管理,如果超过一定时间后还没有触发'disconnect'事件,将会杀死进程。

if (cluster.isMaster) {
  var worker = cluster.fork();
  var timeout;

  worker.on('listening', function(address) {
    worker.send('shutdown');
    worker.disconnect();
    timeout = setTimeout(function() {
      worker.kill();
    }, 2000);
  });

  worker.on('disconnect', function() {
    clearTimeout(timeout);
  });

} else if (cluster.isWorker) {
  var net = require('net');
  var server = net.createServer(function(socket) {
    // connections never end
  });

  server.listen(8000);

  process.on('message', function(msg) {
    if(msg === 'shutdown') {
      // 优雅的关闭服务器所有的连接
    }
  });
}


3.2 是否可连接到主进程:worker.isConnected()

worker.isConnected()
  • 返回值:{Boolean}

当工作进程通过 IPC 通道连接主进程时,返回true,否则 false 。工作进程创建后会连接到主进程。当disconnect事件触发后会关闭连接。


3.3 进程是否结束:worker.isDead()

worker.isDead()
  • 返回值:{Boolean}

当工作进程结束,返回true,否则 false


3.4 结束进程:worker.kill()

worker.kill([signal='SIGTERM'])
  • signal {String},结束进程的信号名

这个方法会杀死工作进程。在主进程里,它会关闭 worker.process,一旦关闭会发送杀死信号。在工作进程里,关闭通道,退出,返回代码0。该方法调用后,.suicide将被设置为true


3.5 发送消息:worker.send()

worker.send(message[, sendHandle][, callback])
  • message {Object}
  • sendHandle {Handle object}
  • callback {Function}
  • 返回值: Boolean

向主进程或工作进程发送消息。

if (cluster.isMaster) {
  var worker = cluster.fork();
  worker.send('hi there');

} else if (cluster.isWorker) {
  process.on('message', function(msg) {
    process.send(msg);
  });
}