Nodejs多线程的探索和实践

本文转载自微信公众号「编程杂技」，作者theanarkh  。转载本文请联系编程杂技公众号。

创新互联成立于2013年，公司以网站建设、做网站、系统开发、网络推广、文化传媒、企业宣传、平面广告设计等为主要业务，适用行业近百种。服务企业客户上千，涉及国内多个省份客户。拥有多年网站建设开发经验。为企业提供专业的网站建设、创意设计、宣传推广等服务。 通过专业的设计、独特的风格，为不同客户提供各种风格的特色服务。

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. 
  
  
  
  
  2.     constructor({ workId, threadId }) {
    3. 
  
  
  
  
  3.         super();
    4. 
  
  
  
  
  4.         this.workId = workId;
    5. 
  
  
  
  
  5.         this.threadId = threadId;
    6. 
  
  
  
  
  6.         workPool[workId] = this;
    7. 
  
  
  
  
  7.     }
    8. 
  
  
  
  
  8. }
    9.

用户提交任务的时候，调度中心返回一个UserWork对象。用户可以使用该对象和调度中心通信。

2.2 调度中心的实现

调度中心的实现大致分为以下几个逻辑。

2.2.1 初始化

 
 
 
 
    
  
  
  
  
  1. constructor(options = {}) {
    2. 
  
  
  
  
  2.        this.options = options;
    3. 
  
  
  
  
  3.        // 线程池总任务数
    4. 
  
  
  
  
  4.        this.totalWork = 0;
    5. 
  
  
  
  
  5.        // 子线程队列
    6. 
  
  
  
  
  6.        this.workerQueue = [];
    7. 
  
  
  
  
  7.        // 核心线程数
    8. 
  
  
  
  
  8.        this.coreThreads = ~~options.coreThreads || config.CORE_THREADS;
    9. 
  
  
  
  
  9.        // 线程池最大线程数，如果不支持动态扩容则最大线程数等于核心线程数
    10. 
  
  
  
  
  10.        this.maxThreads = options.expansion !== false ? Math.max(this.coreThreads, config.MAX_THREADS) : this.coreThreads;
    11. 
  
  
  
  
  11.        // 工作线程处理任务的模式
    12. 
  
  
  
  
  12.        this.sync = options.sync !== false;
    13. 
  
  
  
  
  13.        // 超过任务队列长度时的处理策略
    14. 
  
  
  
  
  14.        this.discardPolicy = options.discardPolicy ? options.discardPolicy : DISCARD_POLICY.NOT_DISCARD;
    15. 
  
  
  
  
  15.        // 是否预创建子线程
    16. 
  
  
  
  
  16.        this.preCreate = options.preCreate === true;
    17. 
  
  
  
  
  17.        this.maxIdleTime = ~~options.maxIdleTime || config.MAX_IDLE_TIME;
    18. 
  
  
  
  
  18.        this.pollIntervalTime = ~~options.pollIntervalTime || config.POLL_INTERVAL_TIME;
    19. 
  
  
  
  
  19.        this.maxWork = ~~options.maxWork || config.MAX_WORK;
    20. 
  
  
  
  
  20.        // 是否预创建线程池
    21. 
  
  
  
  
  21.        this.preCreate && this.preCreateThreads();
    22. 
  
  
  
  
  22.    }
    23.

从初始化代码中我们看到线程池大致支持的能力。

  1. 核心线程数
  2. 最大线程数
  3. 过载时的处理策略，和过载的阈值
  4. 子线程空闲退出的时间和轮询任务的时间
  5. 是否预创建线程池
  6. 是否支持动态扩容

核心线程数是任务数没有达到阈值时的工作线程集合。是处理任务的主力军。任务数达到阈值后，如果支持动态扩容(可配置)则会创建新的线程去处理更多的任务。一旦负载变低，线程空闲时间达到阈值则会自动退出。如果扩容的线程数达到阈值，还有新的任务到来，则根据丢弃策略进行相关的处理。

2.2.2 创建线程

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

创建线程主要是调用nodejs提供的模块进行创建。然后监听子线程的退出和message、error事件。如果是异常退出则补充线程。调度中心维护了一个子线程的队列。记录了每个子线程(worker)的实例和任务数。

2.2.3 选择执行任务的线程

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

选择策略目前是选择任务数最少的，本来还支持随机和轮询方式，但是貌似没有什么场景和必要，就去掉了。

2.2.4 暴露提交任务的接口

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

提交任务的函数比较复杂，提交一个任务的时候，调度中心会根据当前的负载情况和线程数，决定对一个任务做如何处理。如果可以处理，则把任务交给选中的子线程。最后给用户返回一个UserWorker对象。

2.3调度中心和子线程的通信数据结构

 
 
 
 
    
  
  
  
  
  1. class Work {
    2. 
  
  
  
  
  2.     constructor({workId, filename, options}) {
    3. 
  
  
  
  
  3.         // 任务id
    4. 
  
  
  
  
  4.         this.workId = workId;
    5. 
  
  
  
  
  5.         // 文件名
    6. 
  
  
  
  
  6.         this.filename = filename;
    7. 
  
  
  
  
  7.         // 处理结果，由用户代码返回
    8. 
  
  
  
  
  8.         this.data = null;
    9. 
  
  
  
  
  9.         // 执行出错
    10. 
  
  
  
  
  10.         this.error = null;
    11. 
  
  
  
  
  11.         // 执行时入参
    12. 
  
  
  
  
  12.         this.options = options;
    13. 
  
  
  
  
  13.     }
    14. 
  
  
  
  
  14. }
    15.

一个任务对应一个id，目前只支持文件的执行模式，后续会支持字符串。

2.4 子线程的实现

子线程的实现主要分为几个部分

2.4.1 监听调度中心分发的命令

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

2.4.2 轮询是否有任务需要处理

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

不断轮询是否有任务需要处理，如果没有并且空闲时间达到阈值则退出。

2.4.3 处理任务

处理任务模式分为同步和异步

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

用户需要导出一个async函数，使用这种方案主要是为了执行时可以给用户传入参数。并且实现同步。处理完后通知调度中心。下面是异步处理方式，子线程不需要同步等待用户的代码结果。

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

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

 
 
 
 
    
  
  
  
  
  1. module.exports = {
    2. 
  
  
  
  
  2.     // 最大的线程数
    3. 
  
  
  
  
  3.     MAX_THREADS: 50,
    4. 
  
  
  
  
  4.     // 线程池最大任务数
    5. 
  
  
  
  
  5.     MAX_WORK: Infinity,
    6. 
  
  
  
  
  6.     // 默认核心线程数
    7. 
  
  
  
  
  7.     CORE_THREADS: 10,
    8. 
  
  
  
  
  8.     // 最大空闲时间
    9. 
  
  
  
  
  9.     MAX_IDLE_TIME: 10 * 60 * 1000,
    10. 
  
  
  
  
  10.     // 子线程轮询时间
    11. 
  
  
  
  
  11.     POLL_INTERVAL_TIME: 10,
    12. 
  
  
  
  
  12. };
    13. 
  
  
  
  
  13. // 丢弃策略
    14. 
  
  
  
  
  14. const DISCARD_POLICY = {
    15. 
  
  
  
  
  15.     // 报错
    16. 
  
  
  
  
  16.     ABORT: 1,
    17. 
  
  
  
  
  17.     // 在主线程里执行
    18. 
  
  
  
  
  18.     CALLER_RUNS: 2,
    19. 
  
  
  
  
  19.     // 丢弃最老的的任务
    20. 
  
  
  
  
  20.     DISCARD_OLDEST: 3,
    21. 
  
  
  
  
  21.     // 丢弃
    22. 
  
  
  
  
  22.     DISCARD: 4,
    23. 
  
  
  
  
  23.     // 不丢弃
    24. 
  
  
  
  
  24.     NOT_DISCARD: 5,
    25. 
  
  
  
  
  25. };
    26.

支持多个类型的线程池

 
 
 
 
    
  
  
  
  
  1. class AsyncThreadPool extends ThreadPool {
    2. 
  
  
  
  
  2.     constructor(options) {
    3. 
  
  
  
  
  3.         super({...options, sync: false});
    4. 
  
  
  
  
  4.     }
    5. 
  
  
  
  
  5. }
    6. 
  
  
  
  
  6. 
  
  
  
  
  7. class SyncThreadPool extends ThreadPool {
    8. 
  
  
  
  
  8.     constructor(options) {
    9. 
  
  
  
  
  9.         super({...options, sync: true});
    10. 
  
  
  
  
  10.     }
    11. 
  
  
  
  
  11. }
    12. 
  
  
  
  
  12. // cpu型任务的线程池，线程数和cpu核数一样，不支持动态扩容
    13. 
  
  
  
  
  13. class CPUThreadPool extends ThreadPool {
    14. 
  
  
  
  
  14.     constructor(options) {
    15. 
  
  
  
  
  15.         super({...options, coreThreads: cores, expansion: false});
    16. 
  
  
  
  
  16.     }
    17. 
  
  
  
  
  17. }
    18. 
  
  
  
  
  18. // 线程池只有一个线程，类似消息队列
    19. 
  
  
  
  
  19. class SingleThreadPool extends ThreadPool {
    20. 
  
  
  
  
  20.     constructor(options) {
    21. 
  
  
  
  
  21.         super({...options, coreThreads: 1, expansion: false });
    22. 
  
  
  
  
  22.     }
    23. 
  
  
  
  
  23. }
    24. 
  
  
  
  
  24. // 线程数固定的线程池，不支持动态扩容线程
    25. 
  
  
  
  
  25. class FixedThreadPool extends ThreadPool {
    26. 
  
  
  
  
  26.     constructor(options) {
    27. 
  
  
  
  
  27.         super({ ...options, expansion: false });
    28. 
  
  
  
  
  28.     }
    29. 
  
  
  
  
  29. }
    30.

这就是线程池的实现，有很多细节还需要思考。下面是一个性能测试的例子。

3 测试

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

在服务器以单线程和多线程的方式执行以上代码，下面是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 ]

我们发现这个数据差别非常明显。并且随着处理时间的增长，性能差距越明显。

网站名称：Nodejs多线程的探索和实践
分享网址：http://www.csdahua.cn/qtweb/news45/99095.html

网站建设、网络推广公司-快上网，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 快上网

成都快上网为您推荐相关内容

App设计知识

分类信息网

营山产后护理服务　　 　营山母婴护理中心　　 　成都商城网站开发　　 　德阳柴油发电机组　　 　企业网站建设　　 　成都网站推广　　 　红光高低压开关厂　　 　网站免费收录　　 　h5页面制作　　 　成都网站营销推广　　 　工程造价　　 　雅安网站建设　　 　景区导示设计制作　　 　青年旅行社　　 　成都广告设计公司　　 　绵阳主机托管　　 　成都发电机租用　　 　微信小程序开发　　 　卡多尼人工智　　 　定制网站制作