本篇文章和大家探討一下Node.js利用多個核心的方法–worker_threads模塊提供的多線程模型，介紹一下Node.js多進程模型中實現(xiàn)共享內(nèi)存的方法。

Node.js 由于其單線程模型的設(shè)計，導(dǎo)致一個Node進程（的主線程）只能利用一個CPU核心，然而現(xiàn)在的機器基本上都是多核的，這造成了嚴(yán)重的性能浪費。通常來說，想要利用到多個核心一般有以下的方法：

• 編寫Node的C++插件擴充線程池，并在JS代碼中將CPU耗時任務(wù)委托給其它線程處理。

• 使用worker_threads模塊提供的多線程模型（尚在實驗階段）。

• 使用child_process 或者 cluster模塊提供的多進程模型，每個進程都是一個獨立的Node.js進程。

從易用、代碼入侵性、穩(wěn)定性的角度來說，多進程模型通常是首要的選擇?！就扑]學(xué)習(xí)：《nodejs 教程》】

Node.js cluster 多進程模型存在的問題

在cluster模塊提供的多進程模型中，每個Node進程都是一個獨立且完整的應(yīng)用進程，有自己的內(nèi)存空間，其它進程無法訪問。因此雖然在項目啟動時，所有Worker進程具有一致的狀態(tài)和行為，但在之后的運行中無法保證其狀態(tài)維持一致。

例如，項目啟動時有兩個Worker進程，進程A和進程B，兩個進程都聲明了變量a=1。但之后項目接收到一個請求，Master進程將其分派給進程A來處理，這個請求將a的值變更為了2，那么此時進程A的內(nèi)存空間中a=2，但是進程B的內(nèi)存空間中a依舊是1。此時如果有個請求讀取a的值，Master進程將這個請求分派給進程A和進程B時讀取到的結(jié)果是不一致的，這就出現(xiàn)了一致性問題。

cluster模塊在設(shè)計時并沒有給出解決方案，而是要求Worker進程是無狀態(tài)的，即程序員在寫代碼時不應(yīng)該允許在處理請求時修改內(nèi)存中的值，以此來保障所有Worker進程的一致性。然而在實踐中總會有各種各樣的情況需要寫內(nèi)存，比如記錄用戶的登錄狀態(tài)等，在許多企業(yè)的實踐中，通常會把這些狀態(tài)數(shù)據(jù)記錄在外部，例如數(shù)據(jù)庫、redis、消息隊列、文件系統(tǒng)等，每次處理有狀態(tài)請求時會讀寫外部存儲空間。

這不失為一種有效的做法，然而這需要額外引入一個外部存儲空間，同時還要自行處理多進程并發(fā)訪問下的一致性問題，自行維護數(shù)據(jù)的生命周期（因為Node進程和維護在外部的數(shù)據(jù)并不是同步創(chuàng)建和銷毀的），以及在高并發(fā)訪問情況下的IO性能瓶頸（如果是存儲在數(shù)據(jù)庫等非內(nèi)存環(huán)境中）。其實本質(zhì)上來說，我們只是需要一個可供多個進程共享訪問的空間罷了，并不需要持久化存儲，這段空間的生命周期最好與Node進程強綁定，這樣在使用時能省去不少麻煩。因此跨進程的共享內(nèi)存就成了最適合在這種場景使用的方式。

Node.js 的共享內(nèi)存

很遺憾Node本身并未提供共享內(nèi)存的實現(xiàn)，因此我們可以看看npm倉庫中第三方庫的實現(xiàn)。這些庫有些是通過C++插件擴充Node的函數(shù)實現(xiàn)的，有些是通過Node提供的IPC機制實現(xiàn)的，但很遺憾它們的實現(xiàn)都很簡單，并未提供互斥訪問、對象監(jiān)聽等功能，這使得使用者必須自己小心維護這段共享內(nèi)存，否則就會導(dǎo)致時序問題。

轉(zhuǎn)了一圈下來沒找到我想要的。。。那就算了，我自己寫一個。

共享內(nèi)存的設(shè)計

首先我們必須理清楚到底需要個什么樣的共享內(nèi)存，我是根據(jù)我自身的需求出發(fā)（為了在項目中用它來存儲跨進程訪問的狀態(tài)數(shù)據(jù)），同時兼顧通用性，因此會首先考慮以下幾點：

• 以JS對象為基本單位進行讀寫訪問。

• 能夠進程間互斥訪問，一個進程訪問時，其它進程被阻塞。

• 能夠監(jiān)聽共享內(nèi)存中的對象，當(dāng)對象發(fā)生變化的時候監(jiān)聽的進程能被通知到。

• 在滿足上述條件的前提下，實現(xiàn)方式盡可能簡單。

可以發(fā)現(xiàn)，其實我們并不需要操作系統(tǒng)層面的共享內(nèi)存，只需要能夠多個Node進程能訪問同一個對象就行了，那么就可以在Node本身提供的機制上實現(xiàn)。可以使用Master進程的一段內(nèi)存空間作為共享內(nèi)存空間，Worker進程通過IPC將讀寫請求委托給Master進程，由Master進程進行讀寫，然后再通過IPC將結(jié)果返回給Worker進程。

為了讓共享內(nèi)存的使用方式在Master進程和Worker進程中一致，我們可以將對共享內(nèi)存的操作抽離成一個接口，在Master進程和Worker進程中各自實現(xiàn)這個接口。類圖如下圖所示，用一個SharedMemory類作為抽象接口，在server.js入口文件中聲明該對象。其在Master進程中實例化為Manager對象，在Worker進程中實例化為Worker對象。Manager對象來維護共享內(nèi)存，并處理對共享內(nèi)存的讀寫請求，而Worker對象則將讀寫請求發(fā)送到Master進程。

可以使用Manager類中的一個屬性作為共享內(nèi)存對象，訪問該對象的方式與訪問普通JS對象的方式一致，然后再做一層封裝，只暴露get、set、remove等基本操作，避免該屬性直接被修改。

由于Master進程會優(yōu)先于所有Worker進程創(chuàng)建，因此，可以在Master進程中聲明共享內(nèi)存空間之后再創(chuàng)建Worker進程，以此來保證每個Worker進程創(chuàng)建后都可以立即訪問共享內(nèi)存。

為了使用簡單，我們可以將SharedMemory設(shè)計成單例，這樣每個進程中就只有一個實例，并可以在import了SharedMemory之后直接使用。

代碼實現(xiàn)

讀寫控制與IPC通信

首先實現(xiàn)對外接口SharedMemory類，這里沒有使用讓Manager和Worker繼承SharedMemory的方式，而是讓SharedMemory在實例化的時候返回一個Manager或Worker的實例，從而實現(xiàn)自動選擇子類。

在Node 16中isPrimary替代了isMaster，這里為了兼容使用了兩種寫法。

// shared-memory.js class SharedMemory {   constructor() {     if (cluster.isMaster || cluster.isPrimary) {       return new Manager();     } else {       return new Worker();     }   } }

Manager負(fù)責(zé)管理共享內(nèi)存空間，我們直接在Manager對象中增加__sharedMemory__屬性，由于其本身也是JS對象，會被納入JS的垃圾回收管理中，因此我們不需要進行內(nèi)存清理、數(shù)據(jù)遷移等操作，使得實現(xiàn)上非常簡潔。之后在__sharedMemory__之中定義set、get、remove等標(biāo)準(zhǔn)操作來提供訪問方式。

我們通過cluster.on('online', callback)來監(jiān)聽worker進程的創(chuàng)建事件，并在創(chuàng)建后立即用worker.on('message', callback)來監(jiān)聽來自worker進程的IPC通信，并把通信消息交給handle函數(shù)處理。

handle函數(shù)的職責(zé)是區(qū)分worker進程是想進行哪種操作，并取出操作的參數(shù)委托給對應(yīng)的set、get、remove函數(shù)（注意不是__sharedMemory__中的set、get、remove）進行處理，并將處理后的結(jié)果返還給worker進程。

// manager.js const cluster = require('cluster');  class Manager {   constructor() {     this.__sharedMemory__ = {       set(key, value) {         this.memory[key] = value;       },       get(key) {         return this.memory[key];       },       remove(key) {         delete this.memory[key];       },       memory: {},     };      // Listen the messages from worker processes.     cluster.on('online', (worker) => {       worker.on('message', (data) => {         this.handle(data, worker);         return false;       });     });   }    handle(data, target) {     const args = data.value ? [data.key, data.value] : [data.key];     this[data.method](...args).then((value) => {       const msg = {         id: data.id, // workerId         uuid: data.uuid, // communicationID         value,       };       target.send(msg);     });   }    set(key, value) {     return new Promise((resolve) => {       this.__sharedMemory__.set(key, value);       resolve('OK');     });   }    get(key) {     return new Promise((resolve) => {       resolve(this.__sharedMemory__.get(key));     });   }    remove(key) {     return new Promise((resolve) => {       this.__sharedMemory__.remove(key);       resolve('OK');     });   } }

Worker自對象創(chuàng)建開始就使用process.on監(jiān)聽來自Master進程的返回消息（畢竟不能等消息發(fā)送出去以后再監(jiān)聽吧，那就來不及了）。至于__getCallbacks__對象的作用一會兒再說。此時Worker對象便創(chuàng)建完成。

之后項目運行到某個地方的時候，如果要訪問共享內(nèi)存，就會調(diào)用Worker的set、get、remove函數(shù)，它們又會調(diào)用handle函數(shù)將消息通過process.send發(fā)送到master進程，同時，將得到返回結(jié)果時要進行的操作記錄在__getCallbacks__中。當(dāng)結(jié)果返回時，會被之前在process.on中的函數(shù)監(jiān)聽到，并從__getCallbacks__中取出對應(yīng)的回調(diào)函數(shù)，并執(zhí)行。

因為訪問共享內(nèi)存的過程中會經(jīng)過IPC，所以必定是異步操作，所以需要記錄回調(diào)函數(shù)，不能實現(xiàn)成同步的方式，不然會阻塞原本的任務(wù)。

// worker.js const cluster = require('cluster'); const { v4: uuid4 } = require('uuid');  class Worker {   constructor() {     this.__getCallbacks__ = {};      process.on('message', (data) => {       const callback = this.__getCallbacks__[data.uuid];       if (callback && typeof callback === 'function') {         callback(data.value);       }       delete this.__getCallbacks__[data.uuid];     });   }    set(key, value) {     return new Promise((resolve) => {       this.handle('set', key, value, () => {         resolve();       });     });   }    get(key) {     return new Promise((resolve) => {       this.handle('get', key, null, (value) => {         resolve(value);       });     });   }    remove(key) {     return new Promise((resolve) => {       this.handle('remove', key, null, () => {         resolve();       });     });   }    handle(method, key, value, callback) {     const uuid = uuid4(); // 每次通信的uuid     process.send({       id: cluster.worker.id,       method,       uuid,       key,       value,     });     this.__getCallbacks__[uuid] = callback;   } }

一次共享內(nèi)存訪問的完整流程是：調(diào)用Worker的set/get/remove函數(shù) -> 調(diào)用Worker的handle函數(shù)，向master進程通信并將回調(diào)函數(shù)記錄在__getCallbacks__ -> master進程監(jiān)聽到來自worker進程的消息 -> 調(diào)用Manager的handle函數(shù) -> 調(diào)用Manager的set/get/remove函數(shù) -> 調(diào)用__sharedMemory__的set/get/remove函數(shù) -> 操作完成返回Manager的set/get/remove函數(shù) -> 操作完成返回handle函數(shù) -> 向worker進程發(fā)送通信消息 -> worker進程監(jiān)聽到來自master進程的消息 -> 從__getCallbacks__中取出回調(diào)函數(shù)并執(zhí)行。

互斥訪問

到目前為止，我們已經(jīng)實現(xiàn)了讀寫共享內(nèi)存，但還沒有結(jié)束，目前的共享內(nèi)存是存在嚴(yán)重安全問題的。因為這個共享內(nèi)存是可以所有進程同時訪問的，然而我們并沒有考慮并發(fā)訪問時的時序問題。我們來看下面這個例子：

進程A和進程B的目的都是將x的值加1，理想情況下最后x的值應(yīng)該是2，可是最后的結(jié)果卻是1。這是因為進程B在t3時刻給x的值加1的時候，使用的是t2時刻讀取出來的x的值，但此時從全局角度來看，這個值已經(jīng)過期了，因為t3時刻x最新的值已經(jīng)被進程A寫為了1，可是進程B無法知道進程外部的變化，所以導(dǎo)致了t4時刻最后寫回的值又覆蓋掉了進程A寫回的值，等于是進程A的行為被覆蓋掉了。

在多線程、多進程和分布式中并發(fā)情況下的數(shù)據(jù)一致性問題是老大難問題了，這里不再展開討論。

為了解決上述問題，我們必須實現(xiàn)進程間互斥訪問某個對象，來避免同時操作一個對象，從而使進程可以進行原子操作，所謂原子操作就是不可被打斷的一小段連續(xù)操作，為此需要引入鎖的概念。由于讀寫均以對象為基本單位，因此鎖的粒度設(shè)置為對象級別。在某一個進程（的某一任務(wù)）獲取了某個對象的鎖之后，其它要獲取鎖的進程（的任務(wù)）會被阻塞，直到鎖被歸還。而要進行寫操作，則必須要先獲取對象的鎖。這樣在獲取到鎖直到鎖被釋放的這段時間里，該對象在共享內(nèi)存中的值不會被其它進程修改，從而導(dǎo)致錯誤。

在Manager的__sharedMemory__中加入locks屬性，用來記錄哪個對象的鎖被拿走了，lockRequestQueues屬性用來記錄被阻塞的任務(wù)（正在等待鎖的任務(wù)）。并增加getLock函數(shù)和releaseLock函數(shù)，用來申請和歸還鎖，以及handleLockRequest函數(shù)，用來使被阻塞的任務(wù)獲得鎖。在申請鎖時，會先將回調(diào)函數(shù)記錄到lockRequestQueues隊尾（因為此時該對象的鎖可能已被拿走），然后再調(diào)用handleLockRequest檢查當(dāng)前鎖是否被拿走，若鎖還在，則讓隊首的任務(wù)獲得鎖。歸還鎖時，先將__sharedMemory__.locks中對應(yīng)的記錄刪掉，然后再調(diào)用handleLockRequest讓隊首的任務(wù)獲得鎖。

// manager.js const { v4: uuid4 } = require('uuid');  class Manager {   constructor() {     this.__sharedMemory__ = {       ...       locks: {},       lockRequestQueues: {},     };   }    getLock(key) {     return new Promise((resolve) => {       this.__sharedMemory__.lockRequestQueues[key] =         this.__sharedMemory__.lockRequestQueues[key] ?? [];       this.__sharedMemory__.lockRequestQueues[key].push(resolve);       this.handleLockRequest(key);     });   }    releaseLock(key, lockId) {     return new Promise((resolve) => {       if (lockId === this.__sharedMemory__.locks[key]) {         delete this.__sharedMemory__.locks[key];         this.handleLockRequest(key);       }       resolve('OK');     });   }    handleLockRequest(key) {     return new Promise((resolve) => {       if (         !this.__sharedMemory__.locks[key] &&         this.__sharedMemory__.lockRequestQueues[key]?.length > 0       ) {         const callback = this.__sharedMemory__.lockRequestQueues[key].shift();         const lockId = uuid4();         this.__sharedMemory__.locks[key] = lockId;         callback(lockId);       }       resolve();     });   }   ... }

在Worker中，則是增加getLock和releaseLock兩個函數(shù)，行為與get、set類似，都是調(diào)用handle函數(shù)。

// worker.js class Worker {   getLock(key) {     return new Promise((resolve) => {       this.handle('getLock', key, null, (value) => {         resolve(value);       });     });   }    releaseLock(key, lockId) {     return new Promise((resolve) => {       this.handle('releaseLock', key, lockId, (value) => {         resolve(value);       });     });   }   ... }

監(jiān)聽對象

有時候我們需要監(jiān)聽某個對象值的變化，在單進程Node應(yīng)用中這很容易做到，只需要重寫對象的set屬性就可以了，然而在多進程共享內(nèi)存中，對象和監(jiān)聽者都不在一個進程中，這只能依賴Manager的實現(xiàn)。這里，我們選擇了經(jīng)典的觀察者模式來實現(xiàn)監(jiān)聽共享內(nèi)存中的對象。

為此，我們先在__sharedMemory__中加入listeners屬性，用來記錄在對象值發(fā)生變化時監(jiān)聽者注冊的回調(diào)函數(shù)。然后增加listen函數(shù)，其將監(jiān)聽回調(diào)函數(shù)記錄到__sharedMemory__.listeners中，這個監(jiān)聽回調(diào)函數(shù)會將變化的值發(fā)送給對應(yīng)的worker進程。最后，在set和remove函數(shù)返回前調(diào)用notifyListener，將所有記錄在__sharedMemory__.listeners中監(jiān)聽該對象的所有函數(shù)取出并調(diào)用。

// manager.js class Manager {   constructor() {     this.__sharedMemory__ = {       ...       listeners: {},     };   }    handle(data, target) {     if (data.method === 'listen') {       this.listen(data.key, (value) => {         const msg = {           isNotified: true,           id: data.id,           uuid: data.uuid,           value,         };         target.send(msg);       });     } else {       ...     }   }    notifyListener(key) {     const listeners = this.__sharedMemory__.listeners[key];     if (listeners?.length > 0) {       Promise.all(         listeners.map(           (callback) =>             new Promise((resolve) => {               callback(this.__sharedMemory__.get(key));               resolve();             })         )       );     }   }    set(key, value) {     return new Promise((resolve) => {       this.__sharedMemory__.set(key, value);       this.notifyListener(key);       resolve('OK');     });   }    remove(key) {     return new Promise((resolve) => {       this.__sharedMemory__.remove(key);       this.notifyListener(key);       resolve('OK');     });   }    listen(key, callback) {     if (typeof callback === 'function') {       this.__sharedMemory__.listeners[key] =         this.__sharedMemory__.listeners[key] ?? [];       this.__sharedMemory__.listeners[key].push(callback);     } else {       throw new Error('a listener must have a callback.');     }   }   ... }

在Worker中由于監(jiān)聽操作與其它操作不一樣，它是一次注冊監(jiān)聽回調(diào)函數(shù)之后對象的值每次變化都會被通知，因此需要在增加一個__getListenerCallbacks__屬性用來記錄監(jiān)聽操作的回調(diào)函數(shù)，與__getCallbacks__不同，它里面的函數(shù)在收到master的回信之后不會刪除。

// worker.js class Worker {   constructor() {     ...     this.__getListenerCallbacks__ = {};      process.on('message', (data) => {       if (data.isNotified) {         const callback = this.__getListenerCallbacks__[data.uuid];         if (callback && typeof callback === 'function') {           callback(data.value);         }       } else {         ...       }     });   }    handle(method, key, value, callback) {     ...     if (method === 'listen') {       this.__getListenerCallbacks__[uuid] = callback;     } else {       this.__getCallbacks__[uuid] = callback;     }   }    listen(key, callback) {     if (typeof callback === 'function') {       this.handle('listen', key, null, callback);     } else {       throw new Error('a listener must have a callback.');     }   }   ... }

LRU緩存

有時候我們需要用用內(nèi)存作為緩存，但多進程中各進程的內(nèi)存空間獨立，不能共享，因此也需要用到共享內(nèi)存。但是如果用共享內(nèi)存中的一個對象作為緩存的話，由于每次IPC都需要傳輸整個緩存對象，會導(dǎo)致緩存對象不能太大（否則序列化和反序列化耗時太長），而且由于寫緩存對象的操作需要加鎖，進一步影響了性能，而原本我們使用緩存就是為了加快訪問速度。其實在使用緩存的時候通常不會做復(fù)雜操作，大多數(shù)時候也不需要保障一致性，因此我們可以在Manager再增加一個共享內(nèi)存__sharedLRUMemory__，其為一個lru-cache實例，并增加getLRU、setLRU、removeLRU函數(shù)，與set、get、remove函數(shù)類似。

// manager.js const LRU = require('lru-cache');  class Manager {   constructor() {     ...     this.defaultLRUOptions = { max: 10000, maxAge: 1000 * 60 * 5 };     this.__sharedLRUMemory__ = new LRU(this.defaultLRUOptions);   }    getLRU(key) {     return new Promise((resolve) => {       resolve(this.__sharedLRUMemory__.get(key));     });   }    setLRU(key, value) {     return new Promise((resolve) => {       this.__sharedLRUMemory__.set(key, value);       resolve('OK');     });   }    removeLRU(key) {     return new Promise((resolve) => {       this.__sharedLRUMemory__.del(key);       resolve('OK');     });   }   ... }

Worker中也增加getLRU、setLRU、removeLRU函數(shù)。

// worker.js class Worker {   getLRU(key) {     return new Promise((resolve) => {       this.handle('getLRU', key, null, (value) => {         resolve(value);       });     });   }    setLRU(key, value) {     return new Promise((resolve) => {       this.handle('setLRU', key, value, () => {         resolve();       });     });   }    removeLRU(key) {     return new Promise((resolve) => {       this.handle('removeLRU', key, null, () => {         resolve();       });     });   }   ... }

共享內(nèi)存的使用方式

目前共享內(nèi)存的實現(xiàn)已發(fā)到npm倉庫（文檔和源代碼在Github倉庫，歡迎pull request和報bug），可以直接通過npm安裝：

npm i cluster-shared-memory

下面的示例包含了基本使用方法：

const cluster = require('cluster'); // 引入模塊時會根據(jù)當(dāng)前進程 master 進程還是 worker 進程自動創(chuàng)建對應(yīng)的 SharedMemory 對象 require('cluster-shared-memory');  if (cluster.isMaster) {   // 在 master 進程中 fork 子進程   for (let i = 0; i < 2; i++) {     cluster.fork();   } } else {   const sharedMemoryController = require('./src/shared-memory');   const obj = {     name: 'Tom',     age: 10,   };      // 寫對象   await sharedMemoryController.set('myObj', obj);      // 讀對象   const myObj = await sharedMemoryController.get('myObj');      // 互斥訪問對象，首先獲得對象的鎖   const lockId = await sharedMemoryController.getLock('myObj');   const newObj = await sharedMemoryController.get('myObj');   newObj.age = newObj.age + 1;   await sharedMemoryController.set('myObj', newObj);   // 操作完之后釋放鎖   await sharedMemoryController.releaseLock('requestTimes', lockId);      // 或者使用 mutex 函數(shù)自動獲取和釋放鎖   await sharedMemoryController.mutex('myObj', async () => {     const newObjM = await sharedMemoryController.get('myObj');     newObjM.age = newObjM.age + 1;     await sharedMemoryController.set('myObj', newObjM);   });      // 監(jiān)聽對象   sharedMemoryController.listen('myObj', (value) => {     console.log(`myObj: ${value}`);   });      //寫LRU緩存   await sharedMemoryController.setLRU('cacheItem', {user: 'Tom'});      // 讀對象   const cacheItem = await sharedMemoryController.getLRU('cacheItem'); }

缺點

這種實現(xiàn)目前尚有幾個缺點：

• 不能使用PM2的自動創(chuàng)建worker進程的功能。

由于PM2會使用自己的cluster模塊的master進程的實現(xiàn)，而我們的共享內(nèi)存模塊需要在master進程維護一個內(nèi)存空間，則不能使用PM2的實現(xiàn)，因此不能使用PM2的自動創(chuàng)建worker進程的功能。

• 傳輸?shù)膶ο蟊仨毧尚蛄谢?，且不能太大?/p>

• 如果使用者在獲取鎖之后忘記釋放，會導(dǎo)致其它進程一直被阻塞，這要求程序員有良好的代碼習(xí)慣。

原文地址：https://juejin.cn/post/6992091006220894215

作者：FinalZJY