實現(xiàn)一個并發(fā)數(shù)可變的 Promise.all 靜態(tài)方法

codeGoogle 發(fā)布于2019-08-22 16:25 / 2099人閱讀

摘要：在這種情況下，馬上可以想到的一個方法是我們可以使用隊列的數(shù)據(jù)結(jié)構(gòu)來實現(xiàn)調(diào)度粒度為的方法。

實現(xiàn)一個并發(fā)數(shù)可變的 Promise.all 靜態(tài)方法

Promise.all (iterable): The all function returns a new promise which is fulfilled with an array of fulfillment values for the passed promises, or rejects with the reason of the first passed promise that rejects. It resolves all elements of the passed iterable to promises as it runs this algorithm.

Promise.all 靜態(tài)方法具有如下特性：

接收一個 Promise 實例的數(shù)組或具有 Iterator 接口的對象

如果元素不是 Promise 對象則使用 Promise.resolve 轉(zhuǎn)成 Promise 對象

如果全部成功狀態(tài)變?yōu)?resolved，返回值將有序組成一個數(shù)組傳給回調(diào)

只要有一個失敗狀態(tài)就變?yōu)?rejected，返回值將直接傳遞給回調(diào)

該方法的返回值是一個新的 Promise 對象

下面是 Promise.all 的簡單用法：

const p1 = Promise.resolve(1);
const p2 = Promise.resolve(2);
const p3 = Promise.resolve(3);

Promise.all([p1, p2, p3])
    .then((results) => {
        console.log(results); // [1, 2, 3]
    });

const p1 = Promise.resolve(1);
const p2 = Promise.reject(2);
const p3 = Promise.resolve(3);

Promise.all([p1, p2, p3])
    .then((results) => {
        console.log(results);
    }).catch((e) => {
        console.log(e); // 2
    });

如何實現(xiàn) Promise.all 靜態(tài)方法

Promise.all = Promise.all || function(promises) {
    // 如果實參不是數(shù)組則報錯返回
    if (!isArray(promises)) {
        throw new TypeError("You must pass an array to all.");
    }

    // 結(jié)果返回一個新的 Promise 實例
    return new Promise(function(resolve, reject) {
        var i = 0,
            result = [],
            len = promises.length,
            count = len

        // 使用閉包記錄數(shù)組執(zhí)行順序
        function resolver(index) {
            return function(value) {
                resolveAll(index, value);
            };
        }

        // 只要有一個失敗狀態(tài)就變?yōu)?rejected
        function rejecter(reason) {
            reject(reason);
        }

        // 如果全部成功狀態(tài)變?yōu)?resolved
        function resolveAll(index, value) {
            result[index] = value;
            if (--count == 0) {
                resolve(result)
            }
        }

        // 遍歷數(shù)組并發(fā)執(zhí)行異步代碼
        for (; i < len; i++) {
            promises[i].then(resolver(i), rejecter);
        }
    });
}

實現(xiàn)一個調(diào)度粒度可變的 Promise.all 靜態(tài)方法

那么回到題目的問題：如何實現(xiàn)一個調(diào)度粒度可變的 Promise.all 靜態(tài)方法呢？這里首先可能會產(chǎn)生一個疑問就是什么叫調(diào)度粒度可變，實際上很簡單：就是給 all 方法增加一個正整數(shù)類型的參數(shù)，用來標識傳入的 Promise 實例數(shù)組中可以并發(fā)執(zhí)行的最大個數(shù)。

舉例如下，聲明三個不同異步時間的 Promise 實例并調(diào)用 all 方法，正常情況下（我們粗暴的認為）其執(zhí)行時間應(yīng)該為200ms（當(dāng)然這是一個錯誤的答案，在這個例子當(dāng)中我們僅考慮并發(fā)邏輯），而在調(diào)度粒度為1的 all 方法中其執(zhí)行時間應(yīng)該為450ms，在調(diào)度粒度為2的 all 方法中其執(zhí)行時間應(yīng)該為250ms，以此類推。

const p1 = new Promise((resolve) => setTimeout(() => resolve(1), 150));
const p2 = new Promise((resolve) => setTimeout(() => resolve(2), 200));
const p3 = new Promise((resolve) => setTimeout(() => resolve(3), 100));
Promise.all([p1, p2, p3]).then((r) => console.log(r));

在這種情況下，馬上可以想到的一個方法是：我們可以使用隊列的數(shù)據(jù)結(jié)構(gòu)來實現(xiàn)調(diào)度粒度為1的 all 方法。

Promise.all1 = (promises) => {
    if (!isArray(promises)) {
        throw new TypeError("You must pass an array to all.");
    }

    return new Promise((resolve, reject) => {
        const _q = [...promises];
        const result = [];

        function resolver(value) {
            result.push(value);
            next();
        }

        function rejecter(reason) {
            reject(reason);
        }

        function next() {
            if (_q.length) {
                _q.shift().then(resolver, rejecter);
            } else {
                resolve(result);
            }
        }

        next();
    });
}

寫到這兒不難發(fā)現(xiàn)，不同調(diào)度粒度實際上是對隊列每次推出的 Promise 實例數(shù)量最大值的約束，以及對返回結(jié)果的順序索引作出緩存，那么把代碼進行簡單的修改即可實現(xiàn)預(yù)期的功能。

Promise.all2 = (promises, concurrent = promises.length) => {
    if (!Array.isArray(promises)) {
        throw new TypeError("You must pass an array to all.");
    }

    if (concurrent < 1) {
        return Promise.reject();
    }

    return new Promise((resolve, reject) => {
        const queue = [...promises];
        const result = [];
        let total = promises.length;
        let count = concurrent;
        let index = 0;

        function resolver(index) {
            return function(value) {
                resolveAll(index, value);
            };
        }

        function resolveAll(index, value) {
            result[index] = value;
            count++;
            total--;
            next();
            
            if (!total) {
                resolve(result);
            }
        }

        function rejecter(reason) {
            reject(reason);
        }

        function next() {
            while (queue.length && count > 0) {
                count--;
                (queue.shift())().then(resolver(index++), rejecter);
            }
        }

        next();
    });
};

當(dāng)然這里有一個比較吊詭的地方！如果我們按照上面的代碼進行 Promise 實例的聲明，那么在執(zhí)行到 all 方法之前它們就已經(jīng)在并發(fā)執(zhí)行，也就不會有“不同調(diào)度粒度”之說，所以為了實現(xiàn)我們預(yù)期的功能需要在 all 方法內(nèi)部把這些 Promise 實例初始化出來。

function promiseFactory(fn) {
    return function() {
        return new Promise(fn);
    };
}

const p1 = promiseFactory((resolve) => setTimeout(() => resolve(1), 1500));
const p2 = promiseFactory((resolve) => setTimeout(() => resolve(2), 2000));
const p3 = promiseFactory((resolve) => setTimeout(() => resolve(3), 1000));
console.time("hello world!");

Promise.all2([p1, p2, p3], 1).then((r) => {
    console.log(r)
    console.timeEnd("hello world!"); // 4500+ms打印結(jié)果
});

Promise.all2([p1, p2, p3], 2).then((r) => {
    console.log(r)
    console.timeEnd("hello world!"); // 2500+ms打印結(jié)果
});

Promise.all2([p1, p2, p3]).then((r) => {
    console.log(r)
    console.timeEnd("hello world!"); // 2000+ms打印結(jié)果
});

為了能看出明顯的區(qū)別我們把定時器的時間延長到秒，執(zhí)行代碼后完美驗證。

至此結(jié)束。