一、前言

在開(kāi)發(fā)過(guò)程中，有時(shí)會(huì)遇到需要控制任務(wù)并發(fā)執(zhí)行數(shù)量的需求。

例如一個(gè)爬蟲(chóng)程序，可以通過(guò)限制其并發(fā)任務(wù)數(shù)量來(lái)降低請(qǐng)求頻率，從而避免由于請(qǐng)求過(guò)于頻繁被封禁問(wèn)題的發(fā)生。

接下來(lái)，本文介紹如何實(shí)現(xiàn)一個(gè)并發(fā)控制器。

二、示例 

const task = timeout => new Promise((resolve) => setTimeout(() => {    
   resolve(timeout);    
 }, timeout))      
 const taskList = [1000, 3000, 200, 1300, 800, 2000];     
 async function startNoConcurrentControl() {    
   console.time(NO_CONCURRENT_CONTROL_LOG);    
   await Promise.all(taskList.map(item => task(item)));   
   console.timeEnd(NO_CONCURRENT_CONTROL_LOG);    
 }    
 startNoConcurrentControl();   

上述示例代碼利用 Promise.all 方法模擬6個(gè)任務(wù)并發(fā)執(zhí)行的場(chǎng)景，執(zhí)行完所有任務(wù)的總耗時(shí)為 3000 毫秒。

下面會(huì)采用該示例來(lái)驗(yàn)證實(shí)現(xiàn)方法的正確性。

三、實(shí)現(xiàn)

由于任務(wù)并發(fā)執(zhí)行的數(shù)量是有限的，那么就需要一種數(shù)據(jù)結(jié)構(gòu)來(lái)管理不斷產(chǎn)生的任務(wù)。

隊(duì)列的**「先進(jìn)先出」特性可以保證任務(wù)并發(fā)執(zhí)行的順序，在 JavaScript 中可以通過(guò)「數(shù)組來(lái)模擬隊(duì)列」**： 

class Queue {    
    constructor() {    
      this._queue = [];    
    }      
    push(value) {    
      return this._queue.push(value);    
    }      
    shift() {    
      return this._queue.shift();    
    }    
    isEmpty() {    
      return this._queue.length === 0;   
    }    
  }   

對(duì)于每一個(gè)任務(wù)，需要管理其執(zhí)行函數(shù)和參數(shù)： 

class DelayedTask {    
    constructor(resolve, fn, args) {    
      this.resolve = resolve;    
      this.fn = fn;    
      this.args = args;    
    }    
  }   

接下來(lái)實(shí)現(xiàn)核心的 TaskPool 類(lèi)，該類(lèi)主要用來(lái)控制任務(wù)的執(zhí)行： 

class TaskPool {    
    constructor(size) {    
      this.size = size;    
      this.queue = new Queue();   
    }      
    addTask(fn, args) {    
      return new Promise((resolve) => {    
        this.queue.push(new DelayedTask(resolve, fn, args));  
        if (this.size) {    
          this.size--;    
          const { resolve: taskResole, fn, args } = this.queue.shift();    
          taskResole(this.runTask(fn, args));    
        }    
      })    
    }    
    pullTask() {    
      if (this.queue.isEmpty()) {    
        return;    
      }    
      if (this.size === 0) {    
        return;    
      }    
      this.size++;    
      const { resolve, fn, args } = this.queue.shift();    
      resolve(this.runTask(fn, args));    
    }   
     runTask(fn, args) {    
      const result = Promise.resolve(fn(...args));    
       result.then(() => {   
        this.size--;    
        this.pullTask();    
      }).catch(() => {    
        this.size--;    
        this.pullTask();  
       })    
       return result;   
    }    
  }   

TaskPool 包含三個(gè)關(guān)鍵方法：

•  addTask: 將新的任務(wù)放入隊(duì)列當(dāng)中，并觸發(fā)任務(wù)池狀態(tài)檢測(cè)，如果當(dāng)前任務(wù)池非滿(mǎn)載狀態(tài)，則從隊(duì)列中取出任務(wù)放入任務(wù)池中執(zhí)行。
•  runTask: 執(zhí)行當(dāng)前任務(wù)，任務(wù)執(zhí)行完成之后，更新任務(wù)池狀態(tài)，此時(shí)觸發(fā)主動(dòng)拉取新任務(wù)的機(jī)制。
•  pullTask: 如果當(dāng)前隊(duì)列不為空，且任務(wù)池不滿(mǎn)載，則主動(dòng)取出隊(duì)列中的任務(wù)執(zhí)行。

接下來(lái)，將前面示例的并發(fā)數(shù)控制為2個(gè)： 

const cc = new ConcurrentControl(2);      
  async function startConcurrentControl() {    
    console.time(CONCURRENT_CONTROL_LOG);    
    await Promise.all(taskList.map(item => cc.addTask(task, [item])))    
    console.timeEnd(CONCURRENT_CONTROL_LOG);    
  }      
  startConcurrentControl();   

執(zhí)行流程如下：

最終執(zhí)行任務(wù)的總耗時(shí)為 5000 毫秒。

四、高階函數(shù)優(yōu)化參數(shù)傳遞 

await Promise.all(taskList.map(item => cc.addTask(task, [item])))   

手動(dòng)傳遞每個(gè)任務(wù)的參數(shù)的方式顯得非常繁瑣，這里可以通過(guò)**「高階函數(shù)實(shí)現(xiàn)參數(shù)的自動(dòng)透?jìng)鳌?*： 

addTask(fn) {   
    return (...args) => {    
      return new Promise((resolve) => {    
        this.queue.push(new DelayedTask(resolve, fn, args));   
         if (this.size) {    
          this.size--;    
          const { resolve: taskResole, fn: taskFn, args: taskArgs } = this.queue.shift();    
          taskResole(this.runTask(taskFn, taskArgs));    
        }    
      })    
    }    
  }   

改造之后的代碼顯得簡(jiǎn)潔了很多： 

await Promise.all(taskList.map(cc.addTask(task)))   

五、優(yōu)化出隊(duì)操作

數(shù)組一般都是基于一塊**「連續(xù)內(nèi)存」**來(lái)存儲(chǔ)，當(dāng)調(diào)用數(shù)組的 shift 方法時(shí)，首先是刪除頭部元素（時(shí)間復(fù)雜度 O(1)），然后需要將未刪除元素左移一位（時(shí)間復(fù)雜度 O(n)），所以 shift 操作的時(shí)間復(fù)雜度為 O(n)。

由于 JavaScript 語(yǔ)言的特性，V8 在實(shí)現(xiàn) JSArray 的時(shí)候給出了一種空間和時(shí)間權(quán)衡的解決方案，在不同的場(chǎng)景下，JSArray 會(huì)在 FixedArray 和 HashTable 兩種模式間切換。

在 hashTable 模式下，shift 操作省去了左移的時(shí)間復(fù)雜度，其時(shí)間復(fù)雜度可以降低為 O(1)，即使如此，shift 仍然是一個(gè)耗時(shí)的操作。

在數(shù)組元素比較多且需要頻繁執(zhí)行 shift 操作的場(chǎng)景下，可以通過(guò) 「reverse + pop」 的方式優(yōu)化。 

const Benchmark = require('benchmark');    
  const suite = new Benchmark.Suite;      
  suite.add('shift', function() {    
    let count = 10;    
    const arr = generateArray(count);    
    while (count--) {    
      arr.shift();    
    }   
  })   
  .add('reverse + pop', function() {    
    let count = 10;    
    const arr = generateArray(count);    
    arr.reverse();    
    while (count--) {    
      arr.pop();    
    }    
  })    
  .on('cycle', function(event) {    
    console.log(String(event.target));    
  })    
  .on('complete', function() {    
    console.log('Fastest is ' + this.filter('fastest').map('name'));    
    console.log('\n')    
  })    
  .run({    
    async: true    
  })   

通過(guò) benchmark.js 跑出的基準(zhǔn)測(cè)試數(shù)據(jù)，可以很容易地看出哪種方式的效率更高：

回顧之前 Queue 類(lèi)的實(shí)現(xiàn)，由于只有一個(gè)數(shù)組來(lái)存儲(chǔ)任務(wù)，直接使用 reverse + pop 的方式，必然會(huì)影響任務(wù)執(zhí)行的次序。

這里就需要引入雙數(shù)組的設(shè)計(jì)，一個(gè)數(shù)組負(fù)責(zé)入隊(duì)操作，一個(gè)數(shù)組負(fù)責(zé)出隊(duì)操作。 

class HighPerformanceQueue {    
   constructor() {    
     this.q1 = []; // 用于 push 數(shù)據(jù)    
     this.q2 = []; // 用于 shift 數(shù)據(jù)    
   }      
   push(value) {    
     return this.q1.push(value);   
   }    
   shift() {    
     let q2 = this.q2;  
      if (q2.length === 0) {    
       const q1 = this.q1;    
       if (q1.length === 0) {    
         return;    
       }    
       this.q1 = q2; // 感謝 @shaonialife 同學(xué)指正    
       q2 = this.q2 = q1.reverse();    
     }    
     return q2.pop();    
   }    
   isEmpty() {    
     if (this.q1.length === 0 && this.q2.length === 0) { 
        return true;   
     }    
     return false;    
   }    
 }   

最后通過(guò)基準(zhǔn)測(cè)試來(lái)驗(yàn)證優(yōu)化的效果：