成人免费xxxxx在线视频软件_久久精品久久久_亚洲国产精品久久久_天天色天天色_亚洲人成一区_欧美一级欧美三级在线观看

Nodejs多線程的探索和實踐

作者：theanarkh
開發 前端
線程池的設計涉及到很多方面，對于純cpu型的任務，線程數和cpu核數要相等才能達到最優的性能，否則過多的線程引起的上下文切換反而會導致性能下降。

 

[[338602]]

本文轉載自微信公眾號「編程雜技」，作者theanarkh  。轉載本文請聯系編程雜技公眾號。

1 背景

需求中有以下場景

1 對稱解密、非對稱解密

2 壓縮、解壓

3 大量文件的增刪改查

4 處理大量的字符串，解析協議

上面的場景都是非常耗時間的，解密、壓縮、文件操作，nodejs使用了內置的線程池支持了異步。但是處理字符串和解析協議是單純消耗cpu的操作。而且nodejs對解密的支持似乎不是很好。我使用了純js的解密庫，所以無法在nodejs主線程里處理。尤其rsa解密，非常耗時間。

所以這時候就要探索解決方案，nodejs提供了多線程的能力。所以自然就選擇了這種方案。但是這只是初步的想法和方案。因為nodejs雖然提供了多線程能力，但是沒有提供一個應用層的線程池。所以如果我們單純地使用多線程，一個請求一個線程，這顯然不現實。我們不得不實現自己的線程池。本文分享的內容是這個線程池的實現。

線程池的設計涉及到很多方面，對于純cpu型的任務，線程數和cpu核數要相等才能達到最優的性能，否則過多的線程引起的上下文切換反而會導致性能下降。而對于io型的任務，更多的線程理論上是會更好，因為可以更早地給硬盤發出命令，磁盤會優化并持續地處理請求，想象一下，如果發出一個命令，硬盤處理一個，然后再發下一個命令，再處理一個，這樣顯然效率很低。當然，線程數也不是越多越好。線程過多會引起系統負載過高，過多上下文切換也會帶來性能的下降。下面看一下線程池的實現方案。

2 設計思路

首先根據配置創建多個線程(分為預創建和懶創建)，然后對用戶暴露提交任務的接口，由調度中心負責接收任務，然后根據策略選擇處理該任務的線程。子線程一直在輪詢是否有任務需要處理。處理完通知調度中心。

下面看一下具體的實現

2.1 和用戶通信的數據結構

  1. class UserWork extends EventEmitter { 
  2.     constructor({ workId, threadId }) { 
  3.         super(); 
  4.         this.workId = workId; 
  5.         this.threadId = threadId; 
  6.         workPool[workId] = this; 
  7.     } 
  8. } 

用戶提交任務的時候，調度中心返回一個UserWork對象。用戶可以使用該對象和調度中心通信。

2.2 調度中心的實現

調度中心的實現大致分為以下幾個邏輯。

2.2.1 初始化

  1. constructor(options = {}) { 
  2.        this.options = options; 
  3.        // 線程池總任務數 
  4.        this.totalWork = 0; 
  5.        // 子線程隊列 
  6.        this.workerQueue = []; 
  7.        // 核心線程數 
  8.        this.coreThreads = ~~options.coreThreads || config.CORE_THREADS; 
  9.        // 線程池最大線程數，如果不支持動態擴容則最大線程數等于核心線程數 
  10.        this.maxThreads = options.expansion !== false ? Math.max(this.coreThreads, config.MAX_THREADS) : this.coreThreads; 
  11.        // 工作線程處理任務的模式 
  12.        this.sync = options.sync !== false; 
  13.        // 超過任務隊列長度時的處理策略 
  14.        this.discardPolicy = options.discardPolicy ? options.discardPolicy : DISCARD_POLICY.NOT_DISCARD; 
  15.        // 是否預創建子線程 
  16.        this.preCreate = options.preCreate === true; 
  17.        this.maxIdleTime = ~~options.maxIdleTime || config.MAX_IDLE_TIME; 
  18.        this.pollIntervalTime = ~~options.pollIntervalTime || config.POLL_INTERVAL_TIME; 
  19.        this.maxWork = ~~options.maxWork || config.MAX_WORK; 
  20.        // 是否預創建線程池 
  21.        this.preCreate && this.preCreateThreads(); 
  22.    } 

從初始化代碼中我們看到線程池大致支持的能力。

  1. 核心線程數
  2. 最大線程數
  3. 過載時的處理策略，和過載的閾值
  4. 子線程空閑退出的時間和輪詢任務的時間
  5. 是否預創建線程池
  6. 是否支持動態擴容

核心線程數是任務數沒有達到閾值時的工作線程集合。是處理任務的主力軍。任務數達到閾值后，如果支持動態擴容(可配置)則會創建新的線程去處理更多的任務。一旦負載變低，線程空閑時間達到閾值則會自動退出。如果擴容的線程數達到閾值，還有新的任務到來，則根據丟棄策略進行相關的處理。

2.2.2 創建線程

  1. newThread() { 
  2.         let { sync } = this; 
  3.         const worker = new Worker(workerPath, {workerData: { sync, maxIdleTime: this.maxIdleTime, pollIntervalTime: this.pollIntervalTime, }}); 
  4.         const node = { 
  5.             worker, 
  6.             // 該線程處理的任務數量 
  7.             queueLength: 0, 
  8.         }; 
  9.         this.workerQueue.push(node); 
  10.         const threadId = worker.threadId; 
  11.         worker.on('exit', (status) => { 
  12.             // 異常退出則補充線程，正常退出則不補充 
  13.             if (status) { 
  14.                 this.newThread(); 
  15.             } 
  16.             this.totalWork -= node.queueLength; 
  17.             this.workerQueue = this.workerQueue.filter((worker) => { 
  18.                 return worker.threadId !== threadId; 
  19.             }); 
  20.         }); 
  21.         // 和子線程通信 
  22.         worker.on('message', (result) => { 
  23.             const { 
  24.                 work, 
  25.                 event, 
  26.             } = result; 
  27.             const { data, error, workId } = work; 
  28.             // 通過workId拿到對應的userWorker 
  29.             const userWorker = workPool[workId]; 
  30.             delete workPool[workId]; 
  31.             // 任務數減一 
  32.             node.queueLength--; 
  33.             this.totalWork--; 
  34.             switch(event) { 
  35.                 case 'done': 
  36.                     // 通知用戶，任務完成 
  37.                     userWorker.emit('done', data); 
  38.                     break; 
  39.                 case 'error': 
  40.                     // 通知用戶，任務出錯 
  41.                     if (EventEmitter.listenerCount(userWorker, 'error')) { 
  42.                         userWorker.emit('error', error); 
  43.                     } 
  44.                     break; 
  45.                 default: break; 
  46.             } 
  47.         }); 
  48.         worker.on('error', (...rest) => { 
  49.             console.log(...rest) 
  50.         }); 
  51.         return node; 
  52.     } 

創建線程主要是調用nodejs提供的模塊進行創建。然后監聽子線程的退出和message、error事件。如果是異常退出則補充線程。調度中心維護了一個子線程的隊列。記錄了每個子線程(worker)的實例和任務數。

2.2.3 選擇執行任務的線程

  1. selectThead() { 
  2.         let min = Number.MAX_SAFE_INTEGER; 
  3.         let i = 0; 
  4.         let index = 0; 
  5.         // 找出任務數最少的線程，把任務交給他 
  6.         for (; i < this.workerQueue.length; i++) { 
  7.             const { queueLength } = this.workerQueue[i]; 
  8.             if (queueLength < min) { 
  9.                 index = i; 
  10.                 min = queueLength; 
  11.             } 
  12.         } 
  13.         return this.workerQueue[index]; 
  14.     } 

選擇策略目前是選擇任務數最少的，本來還支持隨機和輪詢方式，但是貌似沒有什么場景和必要，就去掉了。

2.2.4 暴露提交任務的接口

  1. submit(filename, options = {}) { 
  2.         return new Promise(async (resolve, reject) => { 
  3.             let thread; 
  4.             // 沒有線程則創建一個 
  5.             if (this.workerQueue.length) { 
  6.                 thread = this.selectThead(); 
  7.                 // 任務隊列非空 
  8.                 if (thread.queueLength !== 0) { 
  9.                     // 子線程個數還沒有達到核心線程數，則新建線程處理 
  10.                     if (this.workerQueue.length < this.coreThreads) { 
  11.                         thread = this.newThread(); 
  12.                     } else if (this.totalWork + 1 > this.maxWork){ 
  13.                         // 總任務數已達到閾值，還沒有達到線程數閾值，則創建 
  14.                         if(this.workerQueue.length < this.maxThreads) { 
  15.                             thread = this.newThread(); 
  16.                         } else { 
  17.                             // 處理溢出的任務 
  18.                             switch(this.discardPolicy) { 
  19.                                 case DISCARD_POLICY.ABORT:  
  20.                                     return reject(new Error('queue overflow')); 
  21.                                 case DISCARD_POLICY.CALLER_RUNS:  
  22.                                     const userWork =  new UserWork({workId: this.generateWorkId(), threadId});  
  23.                                     try { 
  24.                                         const asyncFunction = require(filename); 
  25.                                         if (!isAsyncFunction(asyncFunction)) { 
  26.                                             return reject(new Error('need export a async function')); 
  27.                                         } 
  28.                                         const result = await asyncFunction(options); 
  29.                                         resolve(userWork); 
  30.                                         setImmediate(() => { 
  31.                                             userWork.emit('done', result); 
  32.                                         }); 
  33.                                     } catch (error) { 
  34.                                         resolve(userWork); 
  35.                                         setImmediate(() => { 
  36.                                             userWork.emit('error', error); 
  37.                                         }); 
  38.                                     } 
  39.                                     return; 
  40.                                 case DISCARD_POLICY.DISCARD_OLDEST:  
  41.                                     thread.worker.postMessage({cmd: 'delete'}); 
  42.                                     break; 
  43.                                 case DISCARD_POLICY.DISCARD: 
  44.                                     return reject(new Error('discard')); 
  45.                                 case DISCARD_POLICY.NOT_DISCARD: 
  46.                                     break; 
  47.                                 default:  
  48.                                     break; 
  49.                             } 
  50.                         } 
  51.                     } 
  52.                 } 
  53.             } else { 
  54.                 thread = this.newThread(); 
  55.             } 
  56.             // 生成一個任務id 
  57.             const workId = this.generateWorkId(); 
  58.             // 新建一個work，交給對應的子線程 
  59.             const work = new Work({ workId, filename, options }); 
  60.             const userWork = new UserWork({workId, threadId: thread.worker.threadId}); 
  61.             thread.queueLength++; 
  62.             this.totalWork++; 
  63.             thread.worker.postMessage({cmd: 'add', work}); 
  64.             resolve(userWork); 
  65.         }) 
  66.     } 

提交任務的函數比較復雜，提交一個任務的時候，調度中心會根據當前的負載情況和線程數，決定對一個任務做如何處理。如果可以處理，則把任務交給選中的子線程。最后給用戶返回一個UserWorker對象。

2.3調度中心和子線程的通信數據結構

  1. class Work { 
  2.     constructor({workId, filename, options}) { 
  3.         // 任務id 
  4.         this.workId = workId; 
  5.         // 文件名 
  6.         this.filename = filename; 
  7.         // 處理結果，由用戶代碼返回 
  8.         this.data = null; 
  9.         // 執行出錯 
  10.         this.error = null; 
  11.         // 執行時入參 
  12.         this.options = options; 
  13.     } 
  14. } 

一個任務對應一個id，目前只支持文件的執行模式，后續會支持字符串。

2.4 子線程的實現

子線程的實現主要分為幾個部分

2.4.1 監聽調度中心分發的命令

  1. parentPort.on('message', ({cmd, work}) => { 
  2.     switch(cmd) { 
  3.         case 'delete': 
  4.             return queue.shift(); 
  5.         case 'add': 
  6.             return queue.push(work); 
  7.     } 
  8. }); 

2.4.2 輪詢是否有任務需要處理

  1. function poll() { 
  2.     const now = Date.now(); 
  3.     if (now - lastWorkTime > maxIdleTime && !queue.length) { 
  4.         process.exit(0); 
  5.     } 
  6.     setTimeout(async () => { 
  7.         // 處理任務 
  8.         poll(); 
  9.     } 
  10.     }, pollIntervalTime); 
  11. } 
  12. // 輪詢判斷是否有任務 
  13. poll(); 

不斷輪詢是否有任務需要處理，如果沒有并且空閑時間達到閾值則退出。

2.4.3 處理任務

處理任務模式分為同步和異步

  1. while(queue.length) { 
  2.           const work = queue.shift(); 
  3.           try { 
  4.               const { filename, options } = work; 
  5.               const asyncFunction = require(filename); 
  6.               if (!isAsyncFunction(asyncFunction)) { 
  7.                   return; 
  8.               } 
  9.               lastWorkTime = now; 
  10.  
  11.               const result = await asyncFunction(options); 
  12.               work.data = result; 
  13.               parentPort.postMessage({event: 'done', work}); 
  14.           } catch (error) { 
  15.               work.error = error.toString(); 
  16.               parentPort.postMessage({event: 'error', work}); 
  17.           } 
  18.       } 

用戶需要導出一個async函數，使用這種方案主要是為了執行時可以給用戶傳入參數。并且實現同步。處理完后通知調度中心。下面是異步處理方式，子線程不需要同步等待用戶的代碼結果。

  1. const arr = []; 
  2.        while(queue.length) { 
  3.            const work = queue.shift(); 
  4.            try { 
  5.                const { filename } = work; 
  6.                const asyncFunction = require(filename); 
  7.                if (!isAsyncFunction(asyncFunction)) { 
  8.                    return; 
  9.                } 
  10.                arr.push({asyncFunction, work}); 
  11.            } catch (error) { 
  12.                work.error = error.toString(); 
  13.                parentPort.postMessage({event: 'error', work}); 
  14.            } 
  15.        } 
  16.        arr.map(async ({asyncFunction, work}) => { 
  17.            try { 
  18.                const { options } = work; 
  19.                lastWorkTime = now; 
  20.                const result = await asyncFunction(options); 
  21.                work.data = result; 
  22.                parentPort.postMessage({event: 'done', work}); 
  23.            } catch (e) { 
  24.                work.error = error.toString(); 
  25.                parentPort.postMessage({event: 'done', work}); 
  26.            } 
  27.        }) 

最后還有一些配置和定制化的功能。

  1. module.exports = { 
  2.     // 最大的線程數 
  3.     MAX_THREADS: 50, 
  4.     // 線程池最大任務數 
  5.     MAX_WORK: Infinity, 
  6.     // 默認核心線程數 
  7.     CORE_THREADS: 10, 
  8.     // 最大空閑時間 
  9.     MAX_IDLE_TIME: 10 * 60 * 1000, 
  10.     // 子線程輪詢時間 
  11.     POLL_INTERVAL_TIME: 10, 
  12. }; 
  13. // 丟棄策略 
  14. const DISCARD_POLICY = { 
  15.     // 報錯 
  16.     ABORT: 1, 
  17.     // 在主線程里執行 
  18.     CALLER_RUNS: 2, 
  19.     // 丟棄最老的的任務 
  20.     DISCARD_OLDEST: 3, 
  21.     // 丟棄 
  22.     DISCARD: 4, 
  23.     // 不丟棄 
  24.     NOT_DISCARD: 5, 
  25. }; 

支持多個類型的線程池

  1. class AsyncThreadPool extends ThreadPool { 
  2.     constructor(options) { 
  3.         super({...options, sync: false}); 
  4.     } 
  5. } 
  6.  
  7. class SyncThreadPool extends ThreadPool { 
  8.     constructor(options) { 
  9.         super({...options, sync: true}); 
  10.     } 
  11. } 
  12. // cpu型任務的線程池，線程數和cpu核數一樣，不支持動態擴容 
  13. class CPUThreadPool extends ThreadPool { 
  14.     constructor(options) { 
  15.         super({...options, coreThreads: cores, expansion: false}); 
  16.     } 
  17. } 
  18. // 線程池只有一個線程，類似消息隊列 
  19. class SingleThreadPool extends ThreadPool { 
  20.     constructor(options) { 
  21.         super({...options, coreThreads: 1, expansion: false }); 
  22.     } 
  23. } 
  24. // 線程數固定的線程池，不支持動態擴容線程 
  25. class FixedThreadPool extends ThreadPool { 
  26.     constructor(options) { 
  27.         super({ ...options, expansion: false }); 
  28.     } 
  29. } 

這就是線程池的實現，有很多細節還需要思考。下面是一個性能測試的例子。

3 測試

  1. const { MAX } = require('./constants'); 
  2. module.exports = async function() { 
  3.     let ret = 0; 
  4.     let i = 0; 
  5.     while(i++ < MAX) { 
  6.         ret++; 
  7.         Buffer.from(String(Math.random())).toString('base64'); 
  8.     } 
  9.     return ret; 
  10. } 

在服務器以單線程和多線程的方式執行以上代碼，下面是MAX為10000和100000時，使用CPUThreadPool類型線程池的性能對比(具體代碼參考https://github.com/theanarkh/nodejs-threadpool)。

10000

單線程 [ 358.35, 490.93, 705.23, 982.6, 1155.72 ]

多線程 [ 379.3, 230.35, 315.52, 429.4, 496.04 ]

100000

單線程 [ 2485.5, 4454.63, 6894.5, 9173.16, 11011.16 ]

多線程 [ 1791.75, 2787.15, 3275.08, 4093.39, 3674.91 ]

我們發現這個數據差別非常明顯。并且隨著處理時間的增長，性能差距越明顯。

 

責任編輯：武曉燕 來源： 編程雜技
Node多線程解密
相關推薦

2023-06-16 08:36:25

多線程編程數據競爭

2009-02-24 08:36:51

多線程線程池網絡服務器

2024-10-10 09:46:18

2024-04-30 12:56:00

多線程.NET

2009-03-12 10:52:43

Java線程多線程

2022-08-04 10:32:04

Redis命令

2020-09-22 12:20:23

前端架構插件

2021-09-11 15:26:23

Java多線程線程池

2013-06-13 13:19:38

多線程

2024-02-27 10:44:58

C#線程后端

2025-05-14 08:20:15

2024-11-27 15:58:49

2024-12-05 12:01:09

2023-06-13 13:39:00

多線程異步編程

2025-03-20 10:50:08

RedisCaffeine緩存監控

2019-10-16 17:07:36

Java服務器架構

2013-05-28 15:35:47

html5多線程

2022-12-15 11:26:44

云原生

2023-02-20 15:29:46

異步編碼多線程

2020-10-07 22:21:13

程序員技術線程
點贊
收藏

51CTO技術棧公眾號

主站蜘蛛池模板： 亚洲电影成人 | 日韩欧美三级在线 | 国产欧美一级 | 国产精品 亚洲一区 | av网站免费观看 | 日韩欧美综合在线视频 | 午夜寂寞网站 | 欧美国产精品 | 久久久久中文字幕 | 欧美亚洲高清 | 色综合成人网 | 久久网站免费视频 | 欧美一级免费片 | 亚洲91 | 久久99视频免费观看 | 一区二区在线看 | 精品久久久久一区二区国产 | 欧美一区免费 | 粉色午夜视频 | 亚洲欧美另类在线观看 | 91精品国产91久久久久久 | 99精品99| 久久高清| 97久久精品午夜一区二区 | 99日韩 | 超碰欧美| 一区二区三区精品视频 | 一区二区三区视频在线观看 | 古装人性做爰av网站 | 一区观看 | 国产亚洲精品一区二区三区 | 亚洲精品一区二区网址 | 亚洲一区在线免费观看 | 91一区二区三区在线观看 | 成人免费网视频 | 美日韩免费视频 | 中文字幕乱码一区二区三区 | 久久久精彩视频 | 久久精品黄色 | 一区二区蜜桃 | 国产精品污www一区二区三区 |