前言

上節(jié)我們講到，通過(guò) fork() 或者其他API，創(chuàng)建子進(jìn)程之后，可以通過(guò) send() 和 process.on('message') 進(jìn)行父子進(jìn)程間的通信。這樣就實(shí)現(xiàn)了主進(jìn)程代理請(qǐng)求到工作進(jìn)程，實(shí)現(xiàn)了 Nodejs集群：

父子進(jìn)程間通信

負(fù)載均衡

通過(guò)代理，可以避免端口不能重復(fù)監(jiān)聽(tīng)的問(wèn)題，甚至可以在代理進(jìn)程上做適當(dāng)?shù)?strong>負(fù)載均衡，使得每個(gè)子進(jìn)程可以較為均衡地執(zhí)行任務(wù)。下面我們構(gòu)建了一個(gè)簡(jiǎn)單的 Web 服務(wù)器，并實(shí)現(xiàn)在兩個(gè)工作進(jìn)程之間做簡(jiǎn)單的負(fù)載均衡。

主進(jìn)程，負(fù)責(zé)代理到對(duì)應(yīng)進(jìn)程中：

// main.js
const { fork } = require('child_process');
const normal = fork('subprocess.js', ['normal']);
const special = fork('subprocess.js', ['special']);
// Open up the server and send sockets to child. Use pauseOnConnect to prevent
// 套接字在發(fā)送給子進(jìn)程之前不會(huì)被讀取
const server = require('net').createServer({ pauseOnConnect: true });
let flag = 0;
server.on('connection', (socket) => {
  flag++;
  // this is special priority.
  if (flag % 2 === 0) {
    special.send('socket', socket);
    return;
  }
  // This is normal priority.
  normal.send('socket', socket);
});
server.listen(1337);

這是工作進(jìn)程，接收socket對(duì)象并做出響應(yīng)：

// subprocess.js
process.on('message', (m, socket) => {
  if (m === 'socket') {
    // Check that the client socket exists. 
    // It is possible for the socket to be closed between the time it is
    if (socket) {
      // console.log(`Request handled with ${process.argv[2]} priority`);
      socket.end(`Request handled with ${process.argv[2]} priority, running on ${process.pid}`);
    }
  }
});

然后我又編寫(xiě)了一個(gè) Nodejs 腳本，來(lái)發(fā)出十個(gè) HTTP 請(qǐng)求：

const cp = require("child_process");
for (let i = 0; i < 10; i++) {
  cp.exec(`curl --http0.9 "http://127.0.0.1:1337"`, (err, stdout, stderr) => {
    console.log(`finished: ${i}, and received: `, stdout);
  })
}

最后運(yùn)行結(jié)果如下：

句柄傳遞

在使用 send() 方法時(shí)，我們注意到，除了能通過(guò)IPC發(fā)送數(shù)據(jù)外，還能發(fā)送句柄。第二個(gè)可選參數(shù)就是一個(gè)句柄：

child.send(message, [sendHandle]);

?? 句柄是一種可以用來(lái)標(biāo)識(shí)資源的引用，它的內(nèi)部包含了指向?qū)ο蟮奈募枋龇１热缇浔梢杂脕?lái)標(biāo)識(shí)一個(gè)服務(wù)器端socket對(duì)象、一個(gè)客戶(hù)端socket對(duì)象、一個(gè)UDP套接字、一個(gè)管道等。

在主進(jìn)程將句柄發(fā)送給子進(jìn)程之后，工作模型就從主進(jìn)程響應(yīng)用戶(hù)請(qǐng)求變成了子進(jìn)程監(jiān)聽(tīng)用戶(hù)活動(dòng)：

進(jìn)程對(duì)象send()方法可以發(fā)送的句柄類(lèi)型包括如下幾種：

• net.Socket。TCP套接字。
• net.Server。TCP服務(wù)器，任意建立在TCP服務(wù)上的應(yīng)用層服務(wù)都可以享受到它帶來(lái)的好處。
• net.Native。C++層面的TCP套接字或IPC管道。
• dgram.Socket。UDP套接字。
• dgram.Native。C++層面的UDP套接字。

?? 另外要注意，send()方法能發(fā)送消息和句柄并不意味著它能發(fā)送任意對(duì)象，message 參數(shù)和文件句柄都要先通過(guò) JSON.stringfy() 進(jìn)行序列化后再放入IPC通道中：

集群

通過(guò) child_process模塊，我們完成了父子進(jìn)程的創(chuàng)建和通信，已經(jīng)初步搭建了一個(gè)Node集群。還有一些問(wèn)題需要考慮：

• 性能問(wèn)題。
• 多個(gè)工作進(jìn)程的存活狀態(tài)管理。
• 工作進(jìn)程的平滑重啟。
• 配置或者靜態(tài)數(shù)據(jù)的動(dòng)態(tài)重新載入。
• 其他細(xì)節(jié)。

這其中最重要的便是集群的穩(wěn)定性，這決定了該服務(wù)模型能否真正用于實(shí)踐生成中。雖然我們創(chuàng)建了很多工作進(jìn)程，但每個(gè)工作進(jìn)程依然是在單線(xiàn)程上執(zhí)行的，它的穩(wěn)定性還不能得到完全的保障。我們需要建立起一個(gè)健全的機(jī)制來(lái)保障Node應(yīng)用的健壯性。

子進(jìn)程事件

父進(jìn)程能監(jiān)聽(tīng)到的，與子進(jìn)程相關(guān)的事件：

• error：當(dāng)子進(jìn)程無(wú)法被復(fù)制創(chuàng)建、無(wú)法被殺死、無(wú)法發(fā)送消息時(shí)會(huì)觸發(fā)該事件。
• exit：子進(jìn)程退出時(shí)觸發(fā)該事件。如果是正常退出，這個(gè)事件的第一個(gè)參數(shù)為退出碼，否則為null。如果進(jìn)程是通過(guò)kill()方法被殺死的，會(huì)得到第二個(gè)參數(shù)，它表示殺死進(jìn)程時(shí)的信號(hào)。
• close：在子進(jìn)程的標(biāo)準(zhǔn)輸入輸出流中止時(shí)觸發(fā)該事件，參數(shù)與exit相同。
• disconnect：在父進(jìn)程或子進(jìn)程中調(diào)用disconnect()方法時(shí)觸發(fā)該事件，在調(diào)用該方法時(shí)將關(guān)閉監(jiān)聽(tīng)I(yíng)PC通道。

除了 send() 外，還能通過(guò) kill() 方法給子進(jìn)程發(fā)送消息。kill() 方法并不能真正地將通過(guò)IPC相連的子進(jìn)程殺死，它只是給子進(jìn)程發(fā)送了一個(gè)系統(tǒng)信號(hào)。默認(rèn)情況下，父進(jìn)程將通過(guò) kill() 方法給子進(jìn)程發(fā)送一個(gè) SIGTERM信號(hào)。

// 子進(jìn)程
child.kill([signal]);
// 當(dāng)前進(jìn)程
process.kill(pid, [signal]);
// 監(jiān)聽(tīng)
process.on(signal, callback)

?? 在POSIX標(biāo)準(zhǔn)中，有一套完備的信號(hào)系統(tǒng)，在命令行中執(zhí)行kill -l可以看到詳細(xì)的信號(hào)列表，如下所示：

而 Node 提供了這些信號(hào)對(duì)應(yīng)的信號(hào)事件，每個(gè)進(jìn)程都可以監(jiān)聽(tīng)這些信號(hào)事件。這些信號(hào)事件是用來(lái)通知進(jìn)程的，每個(gè)信號(hào)事件有不同的含義，進(jìn)程在收到響應(yīng)信號(hào)時(shí)，應(yīng)當(dāng)做出約定的行為：

process.on('SIGTERM', () => {
    console.log("got sigterm, exiting...");
    process.exit(1);
});
console.log("process running on: ", process.pid);
process.kill(process.pid, "SIGTERM");

自動(dòng)重啟

有了父子進(jìn)程之間的相關(guān)事件之后，就可以在這些關(guān)系之間創(chuàng)建出需要的機(jī)制了，至少我們能夠通過(guò)監(jiān)聽(tīng)子進(jìn)程的 exit事件 來(lái)獲知其退出的信息。接著前文的多進(jìn)程架構(gòu)，我們?cè)谥鬟M(jìn)程上要加入一些子進(jìn)程管理的機(jī)制，比如重新啟動(dòng)一個(gè)工作進(jìn)程來(lái)繼續(xù)服務(wù)：

主進(jìn)程代碼：

// master.js
// master.js
const { fork } = require('child_process');
const cpus = require('os').cpus();
const server = require('net').createServer();
server.listen(1337);
const workers = {};
// process.on('uncaughtException', function (err) {
//   console.log(`Master uncaughtException:\r\n`);
//   console.log(err);
// });
const createWorker = () => {
  const worker = fork('./worker.js');
  // 收到信號(hào)后立即重啟新進(jìn)程
  worker.on('message', function (message) {
    if (message.act === 'suicide') {
      createWorker();
    }
  });
  // 某個(gè)進(jìn)程終止時(shí)重新啟動(dòng)新的進(jìn)程
  worker.on('exit', () => {
    console.log('Worker ' + worker.pid + ' exited.');
    delete workers[worker.pid];
    // createWorker();
  });
  // 句柄轉(zhuǎn)發(fā)
  worker.send('server', server);
  workers[worker.pid] = worker;
  console.log('Create worker. pid: ' + worker.pid);
};
for (let i = 0; i < cpus.length; i++) {
  createWorker();
}
// server.close();
// 進(jìn)程自己退出時(shí)，讓所有工作進(jìn)程退出
process.on('exit', () => {
  for (let pid in workers) {
    workers[pid].kill();
  }
});

子進(jìn)程代碼：

// worker.js
const http = require('http');
const server = http.createServer((req, res) => {
  res.writeHead(200, {'Content-Type': 'text/plain'});
  res.end('handled by child, pid is ' + process.pid + '\n');
  // 拋出異常，捕獲后終止進(jìn)程
  throw new Error('throw exception');
});
var worker;
process.on('message', (m, tcp) => {
  if (m === 'server') {
    worker = tcp;
    worker.on('connection', (socket) => {
      server.emit('connection', socket);
    });
  }
});
// 捕獲異常后終止進(jìn)程
process.on('uncaughtException', (err) => {
  // 主動(dòng)發(fā)出信號(hào)，避免等待連接斷開(kāi)時(shí)收到新請(qǐng)求而缺少進(jìn)程無(wú)法響應(yīng)
  process.send({
    act: 'suicide'
  });
  // 停止接收新的連接
  worker.close(function () {
    // 所有已有連接斷開(kāi)后，退出進(jìn)程
    process.exit(1);
  });
  // 避免長(zhǎng)連接請(qǐng)求長(zhǎng)時(shí)間無(wú)法終止，5s后自動(dòng)終止
  setTimeout(() => {
    process.exit(1);
  }, 5000)
});

運(yùn)行父進(jìn)程 master.js，控制臺(tái)中會(huì)打印出開(kāi)啟的進(jìn)程 PID：

在 Linux 中，你可以直接使用 kill -9 [pid] 來(lái)終止進(jìn)程。在 Windows 中，你需要打開(kāi)任務(wù)管理器，找到 node.exe 的進(jìn)程，終止其中某個(gè)。此時(shí)命令行會(huì)顯示該進(jìn)程被終止了，然后重新開(kāi)啟一個(gè)新的進(jìn)程。

當(dāng)然，你也可以使用我們之前寫(xiě)的 run.js 腳本，每發(fā)起一個(gè)請(qǐng)求，子進(jìn)程響應(yīng)請(qǐng)求之后會(huì)拋出一個(gè)異常，異常在捕獲之后會(huì)終止該進(jìn)程。

?? 我們之前寫(xiě)的 run.js 腳本是并行執(zhí)行的，此時(shí)會(huì)存在多個(gè)請(qǐng)求被分配到同一個(gè) socket ，即分配到同一個(gè)進(jìn)程中執(zhí)行。那么就會(huì)存在互斥的問(wèn)題，即某個(gè)請(qǐng)求結(jié)束后就終止該進(jìn)程，導(dǎo)致其他請(qǐng)求無(wú)法獲得響應(yīng)而終止。此時(shí)你需要將 exec 方法改為同步方法：

const cp = require("child_process");
const cpus = require("os").cpus();
const sleep = (delay) => {
  const now = Date.now();
  while (Date.now() - now < delay);
  return;
}
for (let i = 0; i < cpus.length; i++) {
  const out = cp.execSync(`curl --http0.9 "http://127.0.0.1:1337"`);
  sleep(1000);
  console.log(out.toString());
}

該模型一旦有異常出現(xiàn)，主進(jìn)程會(huì)創(chuàng)建新的工作進(jìn)程來(lái)為用戶(hù)服務(wù)，舊的進(jìn)程一旦處理完已有連接就自動(dòng)斷開(kāi)。整個(gè)過(guò)程使得我們的應(yīng)用的穩(wěn)定性和健壯性大大提高：

總結(jié)

至此，我們完成了一個(gè)簡(jiǎn)單的基于父子進(jìn)程通信、具備異常重啟進(jìn)程功能的 Web服務(wù)器 就已經(jīng)搭建完成了。對(duì)于 Nodejs 多進(jìn)程編程你也有了初步的了解。接下來(lái)我們將介紹 cluster模塊，并介紹一下在 Nodejs 中進(jìn)行多線(xiàn)程編程。

以上就是Nodejs搭建多進(jìn)程Web服務(wù)器實(shí)現(xiàn)過(guò)程的詳細(xì)內(nèi)容，更多關(guān)于Nodejs搭建多進(jìn)程Web服務(wù)器的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論