p-limit 是一個控制請求併發數量的庫,他的整體代碼不多,思路挺好的,很有學習價值; 舉例 當我們同時發起多個請求時,一般是這樣做的 Promise.all([ requestFn1, requestFn2, requestFn3 ]).then(res =>{}) 或者 requestFn1( ...
p-limit
是一個控制請求併發數量的庫,他的整體代碼不多,思路挺好的,很有學習價值;
舉例
當我們同時發起多個請求時,一般是這樣做的
Promise.all([
requestFn1,
requestFn2,
requestFn3
]).then(res =>{})
或者
requestFn1()
requestFn2()
requestFn3()
而使用 p-limit 限制併發請求數量是這樣做的:
var limit = pLimit(8); // 設置最大併發數量為 8
var input = [ // Limit函數包裝各個請求
limit(() => fetchSomething('1')),
limit(() => fetchSomething('2')),
limit(() => fetchSomething('3')),
limit(() => fetchSomething('4')),
limit(() => fetchSomething('5')),
limit(() => fetchSomething('6')),
limit(() => fetchSomething('7')),
limit(() => fetchSomething('8')),
];
// 執行請求
Promise.all(input).then(res =>{
console.log(res)
})
上面 input
數組包含了 8
個 limit
函數,每個 limit
函數包含了要發起的請求
當設置最大併發數量為 8
時,上面 8
個請求會同時執行
來看下效果,假設每個請求執行時間為1s
。
var fetchSomething = (str) => {
return new Promise((resolve, reject) => {
setTimeout(() => {
console.log(str)
resolve(str)
}, 1000)
})
}
當設置併發請求數量為 2
時
當設置併發請求數量為 3
時
p-limit 限制併發請求數量本質上是,在內部維護了一個請求隊列;
當請求發起時,先將請求推入隊列,判斷當前執行的請求數量是否小於配置的請求併發數量,如果是則執行當前請求,否則等待正在發起的請求中誰請求完了,再從隊列首部取出一個執行;
源碼(v2.3.0)
pLimit
源碼如下(這個源碼是 v2.3.0 版本的,因為項目中引入的版本比較早。後面會分析從 2.3.0
到最新版本的源碼,看看增加或者改進了什麼):
'use strict';
const pTry = require('p-try');
const pLimit = concurrency => {
// 限製為正整數
if (!((Number.isInteger(concurrency) || concurrency === Infinity) && concurrency > 0)) {
return Promise.reject(new TypeError('Expected `concurrency` to be a number from 1 and up'));
}
const queue = []; // 請求隊列
let activeCount = 0; // 當前併發的數量
const next = () => { // 一個請求完成時執行的回調
activeCount--;
if (queue.length > 0) {
queue.shift()();
}
};
const run = (fn, resolve, ...args) => { // 請求開始執行
activeCount++;
const result = pTry(fn, ...args);
resolve(result); // 將結果傳遞給 generator
result.then(next, next); // 請求執行完調用回調
};
// 將請求加入隊列
const enqueue = (fn, resolve, ...args) => {
if (activeCount < concurrency) {
run(fn, resolve, ...args);
} else {
queue.push(run.bind(null, fn, resolve, ...args));
}
};
const generator = (fn, ...args) => new Promise(resolve => enqueue(fn, resolve, ...args));
// 暴露內部屬性給外界
Object.defineProperties(generator, {
activeCount: {
get: () => activeCount
},
pendingCount: {
get: () => queue.length
},
clearQueue: {
value: () => {
queue.length = 0;
}
}
});
return generator;
};
module.exports = pLimit;
module.exports.default = pLimit;
下麵一一剖析下
1、pLimit
函數整體是一個閉包函數,返回了一個名叫 generator
的函數,由 generator
處理併發邏輯,
generator
返回值必須是 promise
,這樣才能被 Promise.all
捕獲到
const generator = (fn,...args) => new Promise((resolve,reject)=7enqueue(fn,resolve,...args))
2、在 enqueue
函數裡面
// 將請求加入隊列
const enqueue = (fn, resolve, ...args) => {
if (activeCount < concurrency) {
run(fn, resolve, ...args);
} else {
queue.push(run.bind(null, fn, resolve, ...args));
}
};
activeCount
表示正在執行的請求數量,當 activeCount
小於配置的併發數量(concurrency
)時,則可以執行當前的 fn
(執行 run
函數),否則推入請求隊列等待。
3、run
函數接收了三個形參
const run = (fn, resolve, ...args) => { // 請求開始執行
activeCount++;
const result = pTry(fn, ...args);
resolve(result);
result.then(next, next);
};
-
fn
表示執行的請求, -
resolve
由generator
定義並往下傳,一直跟蹤到請求執行完畢後,調用resolve(result);
代表generator
函數fulfilled
-
···args
表示其餘的參數,最終會作為fn
的參數。
4、執行 run
函數時
const run = (fn, resolve, ...args) => { // 請求開始執行
activeCount++; // 請求開始執行,當前請求數量 +1
const result = pTry(fn, ...args);
resolve(result);
result.then(next, next);
};
這裡執行 fn
使用的是 const result = pTry(fn,...args)
, pTry
的作用就是創建一個 promise
包裹的結果,不論 fn
是同步函數還是非同步函數
// pTry 源碼
const pTry = (fn,...args) => new Promise((resolve,reject) => resolve(fn(...args)));
現在 fn
執行(fn(...args)
)完畢並兌現(resolve(fn(...args))
)之後,result
就會兌現。
result
兌現後,generator
的 promise
也就兌現了( resolve(result)
),那麼當前請求 fn 的流程就執行完了。
5、當前請求執行完後,對應的當前正在請求的數量也要減一,activeCount--
const next = () => { // 一個請求完成時執行的回調
activeCount--;
if (queue.length > 0) {
queue.shift()();
}
};
然後繼續從隊列頭部取出請求來執行
6、最後暴露內部屬性給外界
Object.defineProperties(generator, {
activeCount: { // 當前正在請求的數量
get: () => activeCount
},
pendingCount: { // 等待執行的數量
get: () => queue.length
},
clearQueue: {
value: () => {
queue.length = 0;
}
}
});
源碼(v2.3.0)=> 源碼(v6.1.0)
從 v2.3.0 到最新的 v6.1.0 版本中間加了一些改進
1、v3.0.0:始終非同步執行傳進 limit 的函數
在 3.0.0
中,作者將請求入隊放在前面,將 if
判斷語句和請求執行置於微任務中運行;正如源碼註釋中解釋的:因為當 run
函數執行時,activeCount
是非同步更新的,那麼這裡的 if
判斷語句也應該非同步執行才能實時獲取到 activeCount
的值。
這樣一開始批量執行 limit(fn)
時,將會先把這些請求全部放入隊列中,然後再根據條件判斷是否執行請求;
2、v3.0.2:修複傳入的無效併發數引起的錯誤;
將 return Promise.reject
改為了直接 throw
一個錯誤
3、v3.1.0:移除 pTry
的依賴;改善性能;
移除了 pTry
依賴,改為了 async
包裹,上面有提到,pTry
是一個 promise
包裝函數,返回結果是一個 promise
;兩者本質都是一樣;
增加了 yocto-queue
依賴,yocto-queue
是一個隊列數據結構,用隊列代替數組,性能更好;隊列的入隊和出隊操作時間複雜度是 O(1)
,而數組的 shift()
是 O(n)
;
4、v5.0.0:修複上下文傳播問題
引入了 AsyncResource
export const AsyncResource = {
bind(fn, _type, thisArg) {
return fn.bind(thisArg);
}
}
這裡用 AsyncResource.bind()
包裹 run.bind(undefined, fn, resolve, args)
,其實不是太明白為啥加這一層。。。這裡用的到三個參數(fn,resolve,args
)都是通過函數傳參過來的,和 this
沒關係吧,各位知道的可以告知下麽。
相關 issue
在這裡
5、6.0.0:性能優化,主要優化的地方在下麵
移除了 AsyncResource.bind()
,改為使用一個立即執行的 promise
,並將 promise
的 resolve
方法插入隊列,一旦 resolve
完成兌現,調用相應請求;相關 issue
在這裡
6、v6.1.0:允許實時修改併發限制數
改變併發數後立馬再檢測是否可以執行請求;
最後
在上面第4
點的,第5
點中的優化沒太看明白,因為執行請求用的到三個參數(fn,resolve,args
)都是通過函數傳參過來的,看起來 this
沒關係,為啥要進行多層 bind
綁定呢?各位知道的可以不吝賜教下麽。