站長資訊網
        最全最豐富的資訊網站

        淺談Node.js多進程模型中如何實現共享內存(代碼詳解)

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

        淺談Node.js多進程模型中如何實現共享內存(代碼詳解)

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

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

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

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

        從易用、代碼入侵性、穩定性的角度來說,多進程模型通常是首要的選擇。【推薦學習:《nodejs 教程》】

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

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

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

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

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

        Node.js 的共享內存

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

        轉了一圈下來沒找到我想要的。。。那就算了,我自己寫一個。

        共享內存的設計

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

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

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

        • 能夠監聽共享內存中的對象,當對象發生變化的時候監聽的進程能被通知到。

        • 在滿足上述條件的前提下,實現方式盡可能簡單。

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

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

        淺談Node.js多進程模型中如何實現共享內存(代碼詳解)

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

        由于Master進程會優先于所有Worker進程創建,因此,可以在Master進程中聲明共享內存空間之后再創建Worker進程,以此來保證每個Worker進程創建后都可以立即訪問共享內存。

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

        代碼實現

        讀寫控制與IPC通信

        首先實現對外接口SharedMemory類,這里沒有使用讓ManagerWorker繼承SharedMemory的方式,而是讓SharedMemory在實例化的時候返回一個ManagerWorker的實例,從而實現自動選擇子類。

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

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

        Manager負責管理共享內存空間,我們直接在Manager對象中增加__sharedMemory__屬性,由于其本身也是JS對象,會被納入JS的垃圾回收管理中,因此我們不需要進行內存清理、數據遷移等操作,使得實現上非常簡潔。之后在__sharedMemory__之中定義setgetremove等標準操作來提供訪問方式。

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

        handle函數的職責是區分worker進程是想進行哪種操作,并取出操作的參數委托給對應的setgetremove函數(注意不是__sharedMemory__中的setgetremove)進行處理,并將處理后的結果返還給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自對象創建開始就使用process.on監聽來自Master進程的返回消息(畢竟不能等消息發送出去以后再監聽吧,那就來不及了)。至于__getCallbacks__對象的作用一會兒再說。此時Worker對象便創建完成。

        之后項目運行到某個地方的時候,如果要訪問共享內存,就會調用Workersetgetremove函數,它們又會調用handle函數將消息通過process.send發送到master進程,同時,將得到返回結果時要進行的操作記錄在__getCallbacks__中。當結果返回時,會被之前在process.on中的函數監聽到,并從__getCallbacks__中取出對應的回調函數,并執行。

        因為訪問共享內存的過程中會經過IPC,所以必定是異步操作,所以需要記錄回調函數,不能實現成同步的方式,不然會阻塞原本的任務。

        // 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;   } }

        一次共享內存訪問的完整流程是:調用Workerset/get/remove函數 -> 調用Workerhandle函數,向master進程通信并將回調函數記錄在__getCallbacks__ -> master進程監聽到來自worker進程的消息 -> 調用Managerhandle函數 -> 調用Managerset/get/remove函數 -> 調用__sharedMemory__set/get/remove函數 -> 操作完成返回Managerset/get/remove函數 -> 操作完成返回handle函數 -> 向worker進程發送通信消息 -> worker進程監聽到來自master進程的消息 -> 從__getCallbacks__中取出回調函數并執行。

        互斥訪問

        到目前為止,我們已經實現了讀寫共享內存,但還沒有結束,目前的共享內存是存在嚴重安全問題的。因為這個共享內存是可以所有進程同時訪問的,然而我們并沒有考慮并發訪問時的時序問題。我們來看下面這個例子:

        時間 進程A 進程B 共享內存中變量x的值
        t0 0
        t1 讀取x(x=0) 0
        t2 x1=x+1(x1=1) 讀取x(x=0) 0
        t3 將x1的值寫回x x2=x+1(x2=1) 1
        t4 將x2的值寫回x 1

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

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

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

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

        // 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中,則是增加getLockreleaseLock兩個函數,行為與getset類似,都是調用handle函數。

        // 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);       });     });   }   ... }

        監聽對象

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

        淺談Node.js多進程模型中如何實現共享內存(代碼詳解)

        為此,我們先在__sharedMemory__中加入listeners屬性,用來記錄在對象值發生變化時監聽者注冊的回調函數。然后增加listen函數,其將監聽回調函數記錄到__sharedMemory__.listeners中,這個監聽回調函數會將變化的值發送給對應的worker進程。最后,在setremove函數返回前調用notifyListener,將所有記錄在__sharedMemory__.listeners中監聽該對象的所有函數取出并調用。

        // 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中由于監聽操作與其它操作不一樣,它是一次注冊監聽回調函數之后對象的值每次變化都會被通知,因此需要在增加一個__getListenerCallbacks__屬性用來記錄監聽操作的回調函數,與__getCallbacks__不同,它里面的函數在收到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緩存

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

        // 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中也增加getLRUsetLRUremoveLRU函數。

        // 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();       });     });   }   ... }

        共享內存的使用方式

        目前共享內存的實現已發到npm倉庫(文檔和源代碼在Github倉庫,歡迎pull request和報bug),可以直接通過npm安裝:

        npm i cluster-shared-memory

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

        const cluster = require('cluster'); // 引入模塊時會根據當前進程 master 進程還是 worker 進程自動創建對應的 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 函數自動獲取和釋放鎖   await sharedMemoryController.mutex('myObj', async () => {     const newObjM = await sharedMemoryController.get('myObj');     newObjM.age = newObjM.age + 1;     await sharedMemoryController.set('myObj', newObjM);   });      // 監聽對象   sharedMemoryController.listen('myObj', (value) => {     console.log(`myObj: ${value}`);   });      //寫LRU緩存   await sharedMemoryController.setLRU('cacheItem', {user: 'Tom'});      // 讀對象   const cacheItem = await sharedMemoryController.getLRU('cacheItem'); }

        缺點

        這種實現目前尚有幾個缺點:

        • 不能使用PM2的自動創建worker進程的功能。

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

        • 傳輸的對象必須可序列化,且不能太大。

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

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

        作者:FinalZJY

        贊(0)
        分享到: 更多 (0)
        網站地圖   滬ICP備18035694號-2    滬公網安備31011702889846號
        主站蜘蛛池模板: 久久亚洲精品中文字幕| 精品久久久久久综合日本| 国产精品三级在线| 下载天堂国产AV成人无码精品网站 | 国产精品白丝AV嫩草影院| 国产精品丝袜一区二区三区| 呦交小u女国产精品视频| 毛片a精品**国产| 国产午夜精品理论片免费观看| 精品国产网红福利在线观看| 国产精品亚洲片在线观看不卡| 热re99久久6国产精品免费| 一本一道久久a久久精品综合 | 国产麻豆一精品一AV一免费| 亚洲日韩国产精品乱| 亚洲精品成人在线| 亚洲A∨精品一区二区三区| 欧美精品黑人粗大视频| 久久se精品一区二区影院| 国产午夜亚洲精品理论片不卡| 丁香色婷婷国产精品视频| 99久久夜色精品国产网站| 日本精品久久久久中文字幕8| 国产精品爱啪在线线免费观看| 国产精品91在线| 久久这里只精品国产99热| 亚洲午夜精品一区二区| 女人香蕉久久**毛片精品| 亚洲国产精品一区二区久久| 91精品国产综合久久香蕉| 亚洲视频精品在线| 国产激情精品一区二区三区| 国产精品1024在线永久免费| 国产麻豆精品入口在线观看| 精品无码久久久久久久动漫| 日本午夜精品理论片A级APP发布| 免费看一级毛片在线观看精品视频 | 亚洲高清国产AV拍精品青青草原| 亚洲麻豆精品国偷自产在线91 | 亚洲精品麻豆av| 亚洲AV无码久久精品蜜桃|