15 行代碼實現併發控制（javascript）

前言

首發於 github blog

作過爬蟲的都知道，要控制爬蟲的請求併發量，其實也就是控制其爬取頻率，以避免被封IP，還有的就是以此來控制爬蟲應用運行內存，不然一會兒處理N個請求，內存分分鐘會爆。

而 python爬蟲通常用多線程來控制併發，

然而若是是node.js爬蟲，因爲其單線程無阻塞性質以及事件循環機制，通常不用多線程來控制併發（固然node.js也能夠實現多線程，此處非重點再也不多講），而是更加簡便地直接在代碼層級上實現併發。

爲圖方便，開發者在開發node爬蟲通常會找一個併發控制的npm包，然而第三方的模塊有時候也並不能徹底知足咱們的特殊需求，這時候咱們可能就須要一個本身定製版的併發控制函數。

下面咱們用15行代碼實現一個併發控制的函數。

具體實現

參數

首先，一個基本的併發控制函數，基本要有如下3個參數：

• list {Array} - 要迭代的數組
• limit {number} - 控制的併發數量
• asyncHandle {function} - 對list的每個項的處理函數

設計

如下以爬蟲爲實例進行講解

設計思路其實很簡單，假如併發量控制是 5

1. 首先，瞬發 5 個異步請求，咱們就獲得了併發的 5 個異步請求

   // limit = 5
   while(limit--) {
       handleFunction(list)
   }

2. 而後，這 5 個異步請求中不管哪個先執行完，都會繼續執行下一個list項

   let recursion = (arr) => {
       return asyncHandle(arr.shift())
           .then(()=>{
               // 迭代數組長度不爲0， 遞歸執行自身
               if (arr.length!==0) return recursion(arr) 
               // 迭代數組長度爲0，結束 
               else return 'finish';
           })
   }

3. 等list全部的項迭代完以後的回調

   return Promise.all(allHandle)

代碼

上述步驟組合起來，就是

/**
 * @params list {Array} - 要迭代的數組
 * @params limit {Number} - 併發數量控制數
 * @params asyncHandle {Function} - 對`list`的每個項的處理函數，參數爲當前處理項，必須 return 一個Promise來肯定是否繼續進行迭代
 * @return {Promise} - 返回一個 Promise 值來確認全部數據是否迭代完成
 */
let mapLimit = (list, limit, asyncHandle) => {
    let recursion = (arr) => {
        return asyncHandle(arr.shift())
            .then(()=>{
                if (arr.length!==0) return recursion(arr)   // 數組還未迭代完，遞歸繼續進行迭代
                else return 'finish';
            })
    };
    
    let listCopy = [].concat(list);
    let asyncList = []; // 正在進行的全部併發異步操做
    while(limit--) {
        asyncList.push( recursion(listCopy) ); 
    }
    return Promise.all(asyncList);  // 全部併發異步操做都完成後，本次併發控制迭代完成
}

測試demo

模擬一下異步的併發狀況

var dataLists = [1,2,3,4,5,6,7,8,9,11,100,123];
var count = 0;
mapLimit(dataLists, 3, (curItem)=>{
    return new Promise(resolve => {
        count++
        setTimeout(()=>{
            console.log(curItem, '當前併發量:', count--)
            resolve();
        }, Math.random() * 5000)  
    });
}).then(response => {
    console.log('finish', response)
})

結果以下：

手動拋出異常中斷併發函數測試：

var dataLists = [1,2,3,4,5,6,7,8,9,11,100,123];
var count = 0;
mapLimit(dataLists, 3, (curItem)=>{
    return new Promise((resolve, reject) => {
        count++
        setTimeout(()=>{
            console.log(curItem, '當前併發量:', count--)
            if(curItem > 4) reject('error happen')
            resolve();
        }, Math.random() * 5000)  
    });
}).then(response => {
    console.log('finish', response)
})

併發控制狀況下，迭代到5,6,7 手動拋出異常，中止後續迭代：