多進程數據處理 · node js

[TOC] >[warning] 說明：本文檔只說明使用頻繁的知識點，有關對應知識的所有具體內容可查看對應的 API 連接 # 1. procss 進程 **`process`對象是一個全局變量，提供有關當前 Node.js 進程的信息并對其進行控制。** **`process` 全部事件與屬性查看官方文檔：[http://nodejs.cn/api/process.html#process\_process](http://nodejs.cn/api/process.html#process_process)** ## 1.1 常用的事件 * ### 1.1.1 beforeExit * node.js 在沒有工作調度時會退出（進程結束），此時會優先觸發 beforeExit 函數 * 監聽 beforeExit 注冊的函數可以異步調用，從而繼續進程（開始新的工作調度） * 監聽函數將傳入 process.exitCode * 顯式終止時不會觸發 beforeExit ~~~ // 監聽函數 process.on('beforeExit', (code) => { console.log('進程 beforeExit 事件的代碼: ', code); }); console.log('此消息最新顯示'); // 顯式終止 process.exit(); 此消息最新顯示進程 exit 事件的代碼: 0 ~~~ * ### 1.1.2 exit * 進程結束時觸發函數 * 監聽函數將傳入 process.exitCode * 監聽函數內只可同步執行，異步函數無效 ~~~ process.on('exit', (code) => { console.log('進程 exit 事件的代碼: ', code); }); process.on('beforeExit', (code) => { console.log('進程 beforeExit 事件的代碼: ', code); }); console.log('此消息最新顯示'); 此消息最新顯示進程 beforeExit 事件的代碼: 0 進程 exit 事件的代碼: 0 ~~~ * ### 1.1.3 message * 如果使用 IPC 通道衍生 Node.js 進程（在子進程與集群會常用），則只要子進程收到父進程的消息，就會觸發事件 * ### 1.1.4 disconnect * 在 IPC 通道關閉時觸發事件，并更改 `process.connected` 為 `false` * 進程調用`process.disconnect()`的效果和父進程調用[`ChildProcess.disconnect()`](http://nodejs.cn/s/ke6TNp)的一樣。 * ### 1.1.5 nextTick * 將回調函數添加到下一個時間點的隊列 ~~~ console.log('開始'); process.nextTick(() => { console.log('下一個時間點的回調'); }); console.log('調度'); 開始調度下一個時間點的回調 ~~~ ## 1.2 常用的屬性 * ### 1.2.1 env * process.env 返回包含用戶環境的對象 * 進程內可 `增刪改` 對象屬性 ~~~ process.env.TEST = 1; console.log(process.env.TEST); // 1 delete process.env.TEST; console.log(process.env.TEST); // => undefined ~~~ * ### 1.2.2 exitCode * 進程退出時的退出碼 * process.exit(code) 可通過 exit 方法指定退出碼 * ### 1.2.3 pid * 進程標識 PID * ### 1.2.4 ppid * 父進程標識 PID # 2. child_procss 子進程 **`child_procss ` 全部事件與屬性查看官方文檔：[http://nodejs.cn/api/child\_process.html#child\_process\_child\_process](http://nodejs.cn/api/child_process.html#child_process_child_process)** ## 2.1 創建異步進程 > 創建異步進程的每一個方法都會返回一個 `ChildProcess` 實例 * ### 2.1.1 fork * 該方法是 `child_process.spawn()` 的一個特例，專門用于衍生新的 Node.js 進程 * 衍生的 Node.js 子進程獨立于父進程，但兩者之間建立的 IPC 通信通道除外 * 通過 fork 創建的子進程可與父進程互發消息 ~~~ const childProcess=require('child_process'); const parent = childProcess.fork('child.js'); parent.send('server', () => { console.log('父子進程建立連接'); }); // child.js process.on('message', function (msg) { if (msg === 'server') { console.log('子進程收到消息'); } }) ~~~ * ### 2.1.2 spawn * 該方法使用給定的`command`衍生一個新進程，并帶上`args`中的命令行參數 ~~~ const childProcess=require('child_process'); const sp = childProcess.spawn('node',['test1.js'],{ cwd:'./test' }); sp.stdout.on('data',function (data) { console.log('子進程標準輸出: '+ data); sp2.stdin.write(data); }); sp.on('exit',function(code, signal) { if (code || code === 0) { console.log('子進程退出，退出代碼為：' + code); } else { console.log('子進程退出，退出信號為：' + signal); } }); ~~~ ## 2.2 創建同步進程 > 同步進程將會阻塞 Node.js 事件循環、暫停任何其他代碼的執行 * ### 2.2.1 spawnSync * spawn 同步版，在子進程完全關閉之前該函數不會返回 ## 2.3 ChildProcess 類 **ChildProcess 實例代表衍生的子進程** **實例通過異步創建進程而生成** ***** **常用事件** * ### 2.3.1 disconnect * 斷開連接后不能再發法消息，且更改 `subprocess.connected` 屬性為`false`，與 `process ` 同名事件使用效果相似 * 調用父進程中的`subprocess.disconnect()`與子進程中的`process.disconnect()`效果一致 * ### 2.3.2 close * 子進程 stdio 流被關閉時觸發 * 與 exit 事件不同，多個進程可能共享同一個 stdio 流 > `options.stdio`選項用于配置在父進程和子進程之間建立的管道 ~~~ const { spawn } = require('child_process'); const ls = spawn('ls', ['-lh', '/usr']); ls.stdout.on('data', (data) => { console.log(`stdout: ${data}`); }); ls.on('close', (code) => { console.log(`子進程使用代碼 ${code} 關閉所有 stdio`); }); ls.on('exit', (code) => { console.log(`子進程使用代碼 ${code} 退出`); }); ~~~ * ### 2.3.3 exit * 進程結束后觸發 * sp.on('exit', (code, single) => {}) code 與 single 兩個值至少一個不為 null * ### 2.3.5 subprocess.send * * ### 2.3.4 message * 子進程使用 process.send() 發送消息時觸發 **常用屬性** * ### 2.3.5 subprocess.channel * 表示子進程的 IPC 通道 * ### 2.3.6 subprocess.connected * 表明是否可以從子進程發送和接收消息 * 為 false 時不可發送或接受消息 * ### 2.3.7 subprocess.stdio * 一個到子進程的管道的稀疏數組，對應于傳給[`child_process.spawn()`](http://nodejs.cn/s/CKoDGf)的被設為`'pipe'`值的[`stdio`](http://nodejs.cn/s/ixmPX9)選項中的位置。`subprocess.stdio[0]`、`subprocess.stdio[1]`和`subprocess.stdio[2]`也分別可用作`subprocess.stdin`、`subprocess.stdout`和`subprocess.stderr`。 * ### 2.3.8 subprocess.stdout * 子進程的`stdout`的可讀流。 * `subprocess.stdout` = `subprocess.stdio[1]` ； `stdio[1]`被設置為`'pipe'`以外的任何值，則該值將會是`null` # 3. cluster 集群 **該模塊可以創建共享服務器端口的子進程。** **單個 Node.js 實例運行在單個線程中，通過啟用一組 Node.js 進程，來充分利用多核系統** **有關集群的全部內容可查看官方文檔：[http://nodejs.cn/api/cluster.html#cluster\_cluster](http://nodejs.cn/api/cluster.html#cluster_cluster)** ~~~ const cluster = require('cluster'); const http = require('http'); const numCPUs = require('os').cpus().length; if (cluster.isMaster) { console.log(`主進程 ${process.pid} 正在運行`); // 衍生工作進程。 for (let i = 0; i < numCPUs; i++) { cluster.fork(); } cluster.on('exit', (worker, code, signal) => { console.log(`工作進程 ${worker.process.pid} 已退出`); }); } else { // 工作進程可以共享任何 TCP 連接。 // 在本例子中，共享的是 HTTP 服務器。 http.createServer((req, res) => { res.writeHead(200); res.end('你好世界\n'); }).listen(8000); console.log(`工作進程 ${process.pid} 已啟動`); } 主進程 2796 正在運行工作進程 14360 已啟動工作進程 12140 已啟動工作進程 6276 已啟動工作進程 14372 已啟動 ~~~ ## 3.1 work 類 > 工作進程由`child_process.fork()`方法創建，因此它們可以使用 IPC 和父進程通信，從而使各進程交替處理連接服務。 **常用事件** * ### 3.1.1 disconnect * 特定于此工作進程 ~~~ cluster.fork().on('disconnect', () => { // 工作進程已斷開連接。 }); ~~~ **常用屬性** * ### 3.1.2 worker.id * 工作進程的標識 * ### 3.1.3 worker.process * 為 child_process.fork() 返回的對象 * 在工作進程中為全局對象 ## 3.2 常用事件 * ### 3.2.1 fork * 新的工作進程被衍生時觸發 * 可用來記錄工作進程活動 > 可以直接調用衍生新的工作進程 ~~~ const timeouts = []; function errorMsg() { console.error('連接出錯'); } cluster.on('fork', (worker) => { timeouts[worker.id] = setTimeout(errorMsg, 2000); }); cluster.on('listening', (worker, address) => { clearTimeout(timeouts[worker.id]); }); cluster.on('exit', (worker, code, signal) => { clearTimeout(timeouts[worker.id]); errorMsg(); }); ~~~ * ### 3.2.2 exit * 任何工作進程關閉時觸發 * 可用于重啟工作進程 ~~~ cluster.on('exit', (worker, code, signal) => { console.log('工作進程 %d 關閉 (%s). 重啟中...', worker.process.pid, signal || code); cluster.fork(); }); ~~~ * ### 3.2.3 online * 衍生工作進后，工作進程運行時觸發 * 主進程衍生工作進程時觸發`'fork'` 而不是 online ~~~ cluster.on('online', (worker) => { console.log('工作進程被衍生后響應'); }); ~~~ ## 3.3 常用屬性 * ### 3.3.1 cluster.isMaster * 該進程為主進程時為 true * ### 3.3.2 cluster.isWorker * 不為主進程時為 true * ### 3.3.3 cluster.worker * 當前工作進程對象 ~~~ const cluster = require('cluster'); if (cluster.isMaster) { console.log('這是主進程'); cluster.fork(); cluster.fork(); } else if (cluster.isWorker) { console.log(`這是工作進程 #${cluster.worker.id}`); } ~~~ * ### 3.3.4 cluster.workers * 存儲所有活躍工作進程對象 * 工作進程斷開連接以及推出后從workers中移除 ~~~ // 遍歷所有工作進程。 function eachWorker(callback) { for (const id in cluster.workers) { callback(cluster.workers[id]); } } eachWorker((worker) => { worker.send('通知所有工作進程'); }); ~~~