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

时间 2019-11-09

标签 node.js node 进程 cluster 处理模块栏目 Node.js 繁體版

原文原文链接

一、process对象

process对象就是处理与进程相关信息的全局对象，不须要require引用，且是EventEmitter的实例。
获取进程信息
process对象提供了不少的API来获取当前进程的运行信息，例如进程运行时间、内存占用、CPU占用、进程号等，具体使用以下所示：javascript

/**
 * 获取当前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，第二个元素是被执行脚本的路径，以下所示：java

var args = process.argv;
if(!args.length){
    process.exit(0);
}else{
    console.log(args.slice(2).join('\n'));
}

执行结果以下：node

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

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

process.on('exit',(code)=>{
    console.log(code);
    setTimeout(()=>console.log(code),1000);//不会执行
});

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

process.on('uncaughtException',(err)=>{
    console.log(err);
});
setTimeout(()=>console.log('nihao'),1000);//1秒后会执行
a();
console.log('hehe');//不会执行

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

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');

执行结果以下：npm

E:\developmentdocument\nodejsdemo>node process-example.js
got the child message：hello parent
got the parent message：hello child!canvas

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

二、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像素处理等。

三、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 }