Node.js：进程、子进程与cluster多核处理模块

2021-08-02 00:46:28

1、process对象

process对象就是处理与进程相关信息的全局对象，不需要require引用，且是EventEmitter的实例。

获取进程信息

process对象提供了很多的API来获取当前进程的运行信息，例如进程运行时间、内存占用、CPU占用、进程号等，具体使用如下所示：

/**

 * 获取当前Node.js进程信息

 */

function getProcessInfo(){

	const memUsage = process.memoryUsage();//内存使用

	const cpuUsage = process.cpuUsage();//cpu使用

	const cfg = process.config;//编译node.js的配置信息

	const env = process.env;//用户环境

	const pwd = process.cwd();//工作目录

	const execPath = process.execPath;//node.exe目录

	const pf = process.platform;//运行nodejs的操作系统平台

	const release = process.release;//nodejs发行版本

	const pid = process.pid;//nodejs进程号

	const arch = process.arch;//运行nodejs的操作系统架构

	const uptime = process.uptime();//nodejs进程运行时间

	return {

		memUsage,

		cpuUsage,

		cfg,

		env,

		pwd,

		execPath,

		pf,

		release,

		pid,

		arch,

		uptime

	}

}

console.log(getProcessInfo());

process.argv获取命令行指令参数

使用node命令执行某个脚本时，可以在指令末尾加上参数，process.argv返回一个数组，第一个元素是process.execPath，第二个元素是被执行脚本的路径，如下所示：

var args = process.argv;

if(!args.length){

	process.exit(0);

}else{

	console.log(args.slice(2).join('\n'));

}

执行结果如下：

E:\developmentdocument\nodejsdemo>node process-example.js a b c

a

b

c

process事件

1、exit事件，当调用process.exit()方法或者事件循环队列没有任何工作时便会触发该事件，监听的回调函数的逻辑必须是同步的，否则不会执行。如下所示：

process.on('exit',(code)=>{

	console.log(code);

	setTimeout(()=>console.log(code),1000);//不会执行

});

2、uncaughtException事件，当一个没有被捕获的异常冒泡到事件队列就会触发该事件，默认打印错误信息并进程退出，当uncaughtException事件有一个以上的 listener 时，会阻止 Node 结束进程。但是这种做法有内存泄露的风险，所以千万不要这么做。如下所示：

process.on('uncaughtException',(err)=>{

	console.log(err);

});

setTimeout(()=>console.log('nihao'),1000);//1秒后会执行

a();

console.log('hehe');//不会执行

3、message事件，进程间使用childProcess.send()方法进行通信，就会触发该事件，使用如下所示：

const cp = require('child_process').fork(`${__dirname}/test.js`);

cp.on('message',(message)=>{

	console.log('got the child message：'+message);

});

cp.send('hello child!');

//test.js

process.on('message',(message)=>{

	console.log('got the parent message：'+message);

});

process.send('hello parent');

执行结果如下：

E:\developmentdocument\nodejsdemo>node process-example.js

got the child message：hello parent

got the parent message：hello child!

process.nextTick方法

将回调函数添加到下一次事件缓存队列中，当前事件循环都执行完毕后，所有的回调函数都会被执行，如下所示：

console.log('hello world');

setTimeout(()=>console.log('settimeout'),10);

process.nextTick(()=>console.log('nexttick'));

console.log('hello nodejs');

执行结果如下所示：

E:\developmentdocument\nodejsdemo>node process-example.js

hello world

hello nodejs

nexttick

settimeout

2、child_process模块

通过child_process模块可以创建子进程，从而实现多进程模式，更好地利用CPU多核计算资源。该模块提供了四种方法创建子进程，分别是child_process.spawn()、child_process.exec()、child_process.execFile()，child_process.fork()，这四个方法都返回一个childProcess对象，该对象实现了EventEmitter的接口，带有stdout，stdin，stderr的对象。

child_process.spawn(command[, args][, options])方法

该方法使用command指令创建一个新进程，参数含义如下：

command，带执行的命令
args，命令行参数数组
options，可选参数，为一个对象

options参数主要拥有以下属性：

cwd，当前工作目录，若没有指定，则使用当前工作目录
env，命令执行环境，默认为process.env
argv0，如果没有指定command，该值会被设置为command
stdio，子进程标准IO配置

返回值为childProcess对象，使用如下所示：

const child_process = require('child_process');

const iconv = require('iconv-lite');

const spawn = child_process.spawn;

const buffArr = [];

let buffLen = 0;

const dirs = spawn('cmd.exe',['/C','dir']);

dirs.stdout.on('data',(data)=>{

	buffArr.push(data);

	buffLen+=data.length;

});

dirs.stderr.on('end',()=>{

	console.log(iconv.decode(Buffer.concat(buffArr,buffLen),'GBK'));

});

dirs.stderr.on('error',(err)=>{

	console.log(err);

});

dirs.on('close',(code)=>{

	console.log(code);

});

执行结果如下：

正在 Ping www.qq.com [14.17.32.211] 具有 32 字节的数据:

来自 14.17.32.211 的回复: 字节=32 时间=2ms TTL=55

来自 14.17.32.211 的回复: 字节=32 时间=2ms TTL=55

来自 14.17.32.211 的回复: 字节=32 时间=3ms TTL=55

来自 14.17.32.211 的回复: 字节=32 时间=3ms TTL=55

14.17.32.211 的 Ping 统计信息:

数据包: 已发送 = 4，已接收 = 4，丢失 = 0 (0% 丢失)，

往返行程的估计时间(以毫秒为单位):

最短 = 2ms，最长 = 3ms，平均 = 2ms

如果输出碰到乱码的时候，可以借助iconv-lite进行转码即可，使用npm install iconv-lite --save。

child_process.exec(command[, options][, callback])方法

新建一个shell执行command指令，并缓存产生的输出结果，方法参数含义如下：

command，待执行的指令，带独立的参数
options，对象，拥有cwd，env，encoding，shell，maxBuffer等属性
callback，回调函数，参数为(error,stdout,stderr)，如果执行成功，error则为null，否则为Error的实例。

返回值也是childProcess对象，该方法与child_process.spawn()方法的区别在于，使用回调函数获得子进程的输出数据，会先将数据缓存在内存中，等待子进程执行完毕之后，再将所有的数据buffer交给回调函数，如果该数据大小超过了maxBuffer(默认为200KB)，则会抛出错误。虽然可以通过参数maxBuffer来设置子进程的缓存大小，但是不建议这么做，因为exec()方法不合适创建返回大量数据的进程，应该就返回一些状态码。

使用如下所示：

exec('netstat /ano | find /C /I "tcp"',(err,stdout,stderr)=>{

	if(err) throw err;

	console.log(stdout);

	console.log(stderr);

});

child_process.execFile(file[, args][, options][, callback])方法

类似与child_process.exec()方法，不同之处是不会创建一个shell，而是直接使用指定的可执行文件创建一个新进程，更有效率，使用如下所示：

execFile('mysql',['--version'],(err,stdout,stderr)=>{

	if(err) throw err;

	console.log(stdout);

	console.log(stderr);

});

child_process.fork(modulePath[, args][, options])方法

创建一个子进程执行module，并与子进程建立IPC通道进行通信，方法返回一个childProcess对象，作为子进程的句柄，通过send()方法向子进程发送信息，监听message事件接收子进程的消息，子进程亦同理。使用如下所示：

const fibonacci = fork('./fibonacci.js');

const n = 10;

fibonacci.on('message',(msg)=>{

	console.log(`fibonacci ${n} is：${msg.result}`);

});

fibonacci.send({n:n});

//fibonacci.js

function fibonacci(n,ac1=1,ac2=1){

	return n<=2?ac2:fibonacci(n-1,ac2,ac1+ac2);

}

process.on('message',(msg)=>{

	process.send({result:fibonacci(msg.n)})

});

child.disconnect()方法

关闭父子进程之间的IPC通道，之后父子进程不能执行通信，并会立即触发disconnect事件，使用如下所示：

const fibonacci = fork('./fibonacci.js');

const n = 10;

fibonacci.on('message',(msg)=>{

	console.log(`fibonacci ${n} is：${msg.result}`);

	fibonacci.disconnect();

});

fibonacci.on('disconnect',()=>{

	console.log('与子进程断开连接.');

});

fibonacci.send({n:n});

//fibonacci.js

function fibonacci(n,ac1=1,ac2=1){

	return n<=2?ac2:fibonacci(n-1,ac2,ac1+ac2);

}

process.on('message',(msg)=>{

	process.send({result:fibonacci(msg.n)})

});

执行结果：

fibonacci 10 is：55

与子进程断开连接.

子进程主要用来做CPU密集型的工作，如fibonacci数列的计算，canvas像素处理等。

3、cluster多核处理模块

Node.js是单线程运行的，不管你的机器有多少个内核，只能用到其中的一个，为了能利用多核计算资源，需要使用多进程来处理应用。cluster模块让我们可以很容易地创建一个负载均衡的集群，自动分配CPU多核资源。

使用如下所示：

const cluster = require('cluster');

const http = require('http');

const cpuNums = require('os').cpus().length;

if(cluster.isMaster){

	for(let i=0;i<cpuNums;i++){

		cluster.fork();

	}

	cluster.on('exit',(worker)=>{

		console.log(`worker${worker.id} exit.`)

	});

	cluster.on('fork',(worker)=>{

		console.log(`fork：worker${worker.id}`)

	});

	cluster.on('listening',(worker,addr)=>{

		console.log(`worker${worker.id} listening on ${addr.address}:${addr.port}`)

	});

	cluster.on('online',(worker)=>{

		console.log(`worker${worker.id} is online now`)

	});

}else{

	http.createServer((req,res)=>{

		console.log(cluster.worker.id);

		res.writeHead(200);

		res.end('hello world');

	}).listen(3000,'127.0.0.1');

}

执行结果：

fork：worker1

fork：worker2

fork：worker3

fork：worker4

worker1 is online now

worker2 is online now

worker3 is online now

worker1 listening on 127.0.0.1:3000

worker4 is online now

worker2 listening on 127.0.0.1:3000

worker3 listening on 127.0.0.1:3000

worker4 listening on 127.0.0.1:3000

cluster工作原理

如上代码所示，master是控制进程，worker是执行进程，每个worker都是使用child_process.fork()函数创建的，因此worker与master之间通过IPC进行通信。

当worker调用用server.listen()方法时会向master进程发送一个消息，让它创建一个服务器socket，做好监听并分享给该worker。如果master已经有监听好的socket，就跳过创建和监听的过程，直接分享。换句话说，所有的worker监听的都是同一个socket，当有新连接进来的时候，由负载均衡算法选出一个worker进行处理。

cluster对象的属性和方法

cluster.isMaster：标志是否master进程，为true则是

cluster.isWorker：标志是否worker进程，为true则是

cluster.worker：获得当前的worker对象，在master进程中使用无效

cluster.workers：获得集群中所有存活的worker对象，子啊worker进程使用无效

cluster.fork()：创建工作进程worker

cluster.disconnect([callback])：断开所有worker进程通信

*cluster对象的事件

Event: 'fork'：监听创建worker进程事件

Event: 'online'：监听worker创建成功事件

Event: 'listening'：监听worker进程进入监听事件

Event: 'disconnect'：监听worker断开事件

Event: 'exit'：监听worker退出事件

Event: 'message'：监听worker进程发送消息事件

使用如下所示：

const cluster = require('cluster');

const http = require('http');

const cpuNums = require('os').cpus().length;

/*process.env.NODE_DEBUG='net';*/

if(cluster.isMaster){

	for(let i=0;i<cpuNums;i++){

		cluster.fork();

	}

	cluster.on('exit',(worker)=>{

		console.log(`worker${worker.id} exit.`)

	});

	cluster.on('fork',(worker)=>{

		console.log(`fork：worker${worker.id}`)

	});

	cluster.on('disconnect',(worker)=>{

		console.log(`worker${worker.id} is disconnected.`)

	});

	cluster.on('listening',(worker,addr)=>{

		console.log(`worker${worker.id} listening on ${addr.address}:${addr.port}`)

	});

	cluster.on('online',(worker)=>{

		console.log(`worker${worker.id} is online now`)

	});

	cluster.on('message',(worker,msg)=>{

		console.log(`got the worker${worker.id}'s msg：${msg}`);

	});

	Object.keys(cluster.workers).forEach((id)=>{

		cluster.workers[id].send(`hello worker${id}`);

	});

}else{

	process.on('message',(msg)=>{

		console.log('worker'+cluster.worker.id+' got the master msg：'+msg);

	});

	process.send('hello master, I am worker'+cluster.worker.id);

	http.createServer((req,res)=>{

		res.writeHead(200);

		res.end('hello world'+cluster.worker.id);

	}).listen(3000,'127.0.0.1');

}

执行结果如下：

fork：worker1

fork：worker2

fork：worker3

fork：worker4

worker1 is online now

worker2 is online now

got the worker1's msg：hello master, I am worker1

worker1 got the master msg：hello worker1

worker1 listening on 127.0.0.1:3000

worker4 is online now

got the worker2's msg：hello master, I am worker2

worker2 got the master msg：hello worker2

worker3 is online now

worker2 listening on 127.0.0.1:3000

got the worker4's msg：hello master, I am worker4

worker4 got the master msg：hello worker4

worker4 listening on 127.0.0.1:3000

got the worker3's msg：hello master, I am worker3

worker3 got the master msg：hello worker3

worker3 listening on 127.0.0.1:3000

在win7环境下，cluster负载均衡情况，如下所示：

服务端代码：

const cluster = require('cluster');

const http = require('http');

const cpuNums = require('os').cpus().length;

if(cluster.isMaster){

	var i = 0;

	const widArr = [];

	for(let i=0;i<cpuNums;i++){

		cluster.fork();

	}

	cluster.on('message',(worker,msg)=>{

		if(msg === 'ex'){

			i++;

			widArr.push(worker.id);

			(i>=80)&&(process.exit(0));

		}

	});

	process.on('exit', (code) => {

		console.log(analyzeArr(widArr));

	});

	//统计每个worker被调用的次数

	function analyzeArr(arr) {

		let obj = {};

		arr.forEach((id, idx, arr) => {

			obj['work' + id] = obj['work' + id] !== void 0 ? obj['work' + id] + 1 : 1;

		});

		return obj;

	}

}else{

	http.createServer((req,res)=>{

		console.log(`worker${cluster.worker.id}`);

		process.send('ex');

		res.writeHead(200);

		res.end('hello world'+cluster.worker.id);

	}).listen(3000,'127.0.0.1');

}

使用Apache的AB命令进行测试，并发40，总共80：C:\Users\learn>ab -c 40 -n 80 http://127.0.0.1:3000/。

测试结果：

{ work4: 19, work3: 20, work1: 19, work2: 22 }

码农公寓

1、process对象

2、child_process模块

3、cluster多核处理模块

相关文章