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

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 }

上一篇:js限制输入框只能输入数字


下一篇:Android studio-约束布局