Promise.all并發(fā)限制

sutaking 發(fā)布于2019-08-23 11:58 / 1838人閱讀

摘要：并發(fā)限制指的是，每個時(shí)刻并發(fā)執(zhí)行的數(shù)量是固定的，最終的執(zhí)行結(jié)果還是保持與原來的一致。換句話說，就是把生成數(shù)組的控制權(quán)，交給并發(fā)控制邏輯。

背景

通常，我們在需要保證代碼在多個異步處理之后執(zhí)行，會用到：

Promise.all(promises: []).then(fun: function);

Promise.all可以保證，promises數(shù)組中所有promise對象都達(dá)到resolve狀態(tài)，才執(zhí)行then回調(diào)。

這時(shí)候考慮一個場景：如果你的promises數(shù)組中每個對象都是http請求，或者說每個對象包含了復(fù)雜的調(diào)用處理。而這樣的對象有幾十萬個。

那么會出現(xiàn)的情況是，你在瞬間發(fā)出幾十萬http請求（tcp連接數(shù)不足可能造成等待），或者堆積了無數(shù)調(diào)用棧導(dǎo)致內(nèi)存溢出。

這時(shí)候，我們就需要考慮對Promise.all做并發(fā)限制。

Promise.all并發(fā)限制指的是，每個時(shí)刻并發(fā)執(zhí)行的promise數(shù)量是固定的，最終的執(zhí)行結(jié)果還是保持與原來的Promise.all一致。

實(shí)現(xiàn)

我們知道，promise并不是因?yàn)檎{(diào)用Promise.all才執(zhí)行，而是在實(shí)例化promise對象的時(shí)候就執(zhí)行了，在理解這一點(diǎn)的基礎(chǔ)上，要實(shí)現(xiàn)并發(fā)限制，只能從promise實(shí)例化上下手。

換句話說，就是把生成promises數(shù)組的控制權(quán)，交給并發(fā)控制邏輯。

這里我并不打算一步步實(shí)現(xiàn)這個這個功能，npm中有很多實(shí)現(xiàn)這個功能的第三方包，比如async-pool、es6-promise-pool、p-limit，這里我直接拿async-pool的代碼來分析一下實(shí)現(xiàn)原理。

代碼很簡單，去掉不必要的代碼，加上自己的注釋，大概內(nèi)容如下：

function asyncPool(poolLimit, array, iteratorFn) {
    let i = 0;
    const ret = [];
    const executing = [];
    const enqueue = function () {
        // 邊界處理，array為空數(shù)組
        if (i === array.length) {
            return Promise.resolve();
        }
        // 每調(diào)一次enqueue，初始化一個promise
        const item = array[i++];
        const p = Promise.resolve().then(() => iteratorFn(item, array));
        // 放入promises數(shù)組
        ret.push(p);
        // promise執(zhí)行完畢，從executing數(shù)組中刪除
        const e = p.then(() => executing.splice(executing.indexOf(e), 1));
        // 插入executing數(shù)字，表示正在執(zhí)行的promise
        executing.push(e);
        // 使用Promise.rece，每當(dāng)executing數(shù)組中promise數(shù)量低于poolLimit，就實(shí)例化新的promise并執(zhí)行
        let r = Promise.resolve();
        if (executing.length >= poolLimit) {
            r = Promise.race(executing);
        }
        // 遞歸，直到遍歷完array
        return r.then(() => enqueue());
    };
    return enqueue().then(() => Promise.all(ret));
}

因?yàn)槭莗romise加上遞歸，所以在代碼注釋上不太好標(biāo)注執(zhí)行順序，但是大概的邏輯可以總結(jié)為：

從array第1個元素開始，初始化promise對象，同時(shí)用一個executing數(shù)組保存正在執(zhí)行的promise

不斷初始化promise，直到達(dá)到poolLimt

使用Promise.race，獲得executing中promise的執(zhí)行情況，當(dāng)有一個promise執(zhí)行完畢，繼續(xù)初始化promise并放入executing中

所有promise都執(zhí)行完了，調(diào)用Promise.all返回

使用方式就是：

const timeout = i => new Promise(resolve => setTimeout(() => resolve(i), i));
return asyncPool(2, [1000, 5000, 3000, 2000], timeout).then(results => {
    ...
});

總結(jié)

所謂promise并發(fā)限制，其實(shí)根源上就是控制promise的實(shí)例化。如果是通過第三方函數(shù)，那么就把創(chuàng)建promise的控制權(quán)交給第三方即可。

然而這樣的實(shí)現(xiàn)效果，本質(zhì)上來說已經(jīng)拋棄了Promise.all而另辟蹊徑。所以期待有一天promise標(biāo)準(zhǔn)能提供這個功能