快捷導(dǎo)航

JS使用Promise控制請求并發(fā)數(shù)

更新時間：2023年05月25日 11:28:29 作者：JetTsang

現(xiàn)在面試過程當(dāng)中，手寫題必然是少不了的，其中碰到比較多的無非就是當(dāng)屬請求并發(fā)控制了，所以本文為大家整理了JS使用Promise控制請求并發(fā)數(shù)的示例代碼，希望對大家有所幫助

前言

現(xiàn)在面試過程當(dāng)中，手寫題必然是少不了的，其中碰到比較多的無非就是當(dāng)屬請求并發(fā)控制了。而基本上前端項目都是通過axios來實現(xiàn)異步請求的封裝，因此這其實是考你對Promise以及異步編程的理解了。

引出

題目：

// 設(shè)計一個函數(shù)，可以限制請求的并發(fā)，同時請求結(jié)束之后，調(diào)用callback函數(shù)
// sendRequest(requestList:,limits,callback):void
sendRequest(
[()=>request('1'),
()=>request('2'),
()=>request('3'),
()=>request('4')],
3, //并發(fā)數(shù)
(res)=>{
    console.log(res)
})
// 其中request 可以是： 
function request (url,time=1){
    return new Promise((resolve,reject)=>{
        setTimeout(()=>{
            console.log('請求結(jié)束：'+url);
            if(Math.random() > 0.5){
                resolve('成功')
            }else{
                reject('錯誤;')
            }
        },time*1e3)
    })
}

明確概念

這里有幾個概念需要明確一下

并發(fā)：并發(fā)是多個任務(wù)同時交替的執(zhí)行（因為cpu執(zhí)行指令的速度非常之快，它可以不必按順序一段代碼一段代碼的執(zhí)行，這樣效率反而更加低下），這樣看起來就是一起執(zhí)行的，所以叫并發(fā)。
并行：可以理解為多個物理cpu或者有分布式系統(tǒng)，是真正的'同時'執(zhí)行
并發(fā)控制：意思是多個并發(fā)的任務(wù)，一旦有任務(wù)完成，就立刻開啟下一個任務(wù)
切片控制：將并發(fā)任務(wù)切片的分配出來，比如10個任務(wù)，切成2個片，每片有5個任務(wù)，當(dāng)前一片的任務(wù)執(zhí)行完畢，再開始下一個片的任務(wù)，這樣明顯效率沒并發(fā)控制那么高了

思路

首先執(zhí)行能執(zhí)行的并發(fā)任務(wù)，根據(jù)并發(fā)的概念，每個任務(wù)執(zhí)行完畢后，撈起下一個要執(zhí)行的任務(wù)。

將關(guān)鍵步驟拆分出合適的函數(shù)來組織代碼

循環(huán)去啟動能執(zhí)行的任務(wù)
取出任務(wù)并且推到執(zhí)行器執(zhí)行
執(zhí)行器內(nèi)更新當(dāng)前的并發(fā)數(shù)，并且觸發(fā)撈起任務(wù)
撈起任務(wù)里面可以觸發(fā)最終的回調(diào)函數(shù)和調(diào)起執(zhí)行器繼續(xù)執(zhí)行任務(wù)

實現(xiàn)

1.定義常量和函數(shù)

function sendRequest(requestList,limits,callback){
    // 定義執(zhí)行隊列，表示所有待執(zhí)行的任務(wù)
    const promises = requestList.slice()
    // 定義開始時能執(zhí)行的并發(fā)數(shù)
    const concurrentNum = Math.min(limits,requestList.length)
    let concurrentCount = 0 // 當(dāng)前并發(fā)數(shù)
    // 啟動初次能執(zhí)行的任務(wù)
    const runTaskNeeded = ()=>{
        let i = 0
        while(i<concurrentNum){
            runTask()
        }
    }
    // 取出任務(wù)并推送到執(zhí)行器
    const runTask = ()=>{}
    // 執(zhí)行器，這里去執(zhí)行任務(wù)
    const runner = ()=>{}
    // 撈起下一個任務(wù)
    const picker = ()=>{}
    // 開始執(zhí)行！
    runTaskNeeded()
}

2.實現(xiàn)對應(yīng)的函數(shù)

function sendRequest(requestList,limits,callback){
    const promises = requestList.slice() // 取得請求list（淺拷貝一份）
    // 得到開始時，能執(zhí)行的并發(fā)數(shù)
    const concurrentNum = Math.min(limits,requestList.length)
    let concurrentCount = 0 // 當(dāng)前并發(fā)數(shù)
    // 第一次先跑起可以并發(fā)的任務(wù)
    const runTaskNeeded = ()=>{
        let i = 0
        // 啟動當(dāng)前能執(zhí)行的任務(wù)
        while(i<concurrentNum){
            i++
            runTask()
        }
    }
    // 取出任務(wù)并且執(zhí)行任務(wù)
    const runTask = ()=>{
        const task = promises.shift()
        task && runner(task)
    }
    // 執(zhí)行器
    // 執(zhí)行任務(wù)，同時更新當(dāng)前并發(fā)數(shù)
    const runner = async (task)=>{
        try {
            concurrentCount++
            await task()
        } catch (error) {
        }finally{
            // 并發(fā)數(shù)--
            concurrentCount--
            // 撈起下一個任務(wù)
            picker()
        }
    }
// 撈起下一個任務(wù)
    const picker = ()=>{
        // 任務(wù)隊列里還有任務(wù)并且此時還有剩余并發(fā)數(shù)的時候 執(zhí)行
        if(concurrentCount < limits && promises.length > 0 ){
            // 繼續(xù)執(zhí)行任務(wù)
            runTask()
        // 隊列為空的時候，并且請求池清空了，就可以執(zhí)行最后的回調(diào)函數(shù)了
        }else if(promises.length ==0 && concurrentCount ==0 ){
            // 執(zhí)行結(jié)束
            callback && callback()
        }
    }
    // 入口執(zhí)行
    runTaskNeeded()
}

另一種實現(xiàn)

核心代碼是判斷是當(dāng)你【有任務(wù)執(zhí)行完成】，再去判斷是否有剩余還有任務(wù)可執(zhí)行。可以先維護(hù)一個pool（代表當(dāng)前執(zhí)行的任務(wù)），利用await Promise.race這個pool，不就知道是否有任務(wù)執(zhí)行完畢了嗎？

async function sendRequest(requestList,limits,callback){
    // 維護(hù)一個promise隊列
    const promises = []
    // 當(dāng)前的并發(fā)池,用Set結(jié)構(gòu)方便刪除
    const pool = new Set() // set也是Iterable<any>[]類型，因此可以放入到race里
    // 開始并發(fā)執(zhí)行所有的任務(wù)
    for(let request of requestList){
        // 開始執(zhí)行前，先await 判斷 當(dāng)前的并發(fā)任務(wù)是否超過限制
        if(pool.size >= limits){
            // 這里因為沒有try catch ，所以要捕獲一下錯誤，不然影響下面微任務(wù)的執(zhí)行
            await Promise.race(pool)
            .catch(err=>err)
        }
        const promise = request()// 拿到promise
        // 刪除請求結(jié)束后，從pool里面移除
        const cb = ()=>{
            pool.delete(promise)
        }
        // 注冊下then的任務(wù)
        promise.then(cb,cb)
        pool.add(promise)
        promises.push(promise)
    }
    // 等最后一個for await 結(jié)束，這里是屬于最后一個 await 后面的 微任務(wù)
    // 注意這里其實是在微任務(wù)當(dāng)中了，當(dāng)前的promises里面是能確保所有的promise都在其中(前提是await那里命中了if)
    Promise.allSettled(promises).then(callback,callback)
}

總結(jié)一下要點：

利用race的特性可以找到并發(fā)任務(wù) 里最快結(jié)束的請求
利用for await 可以保證for結(jié)構(gòu)體下面的代碼是最后await 后的微任務(wù)，而在最后一個微任務(wù)下，可以保證所有的promise已經(jīng)存入promises里（如果沒命中任何一個await，即限制并發(fā)數(shù)>任務(wù)數(shù)的時候，雖然不是在微任務(wù)當(dāng)中，也可以保證所有的promise都在里面），最后利用allSettled，等待所有的promise狀態(tài)轉(zhuǎn)變后，調(diào)用回調(diào)函數(shù)
并發(fā)任務(wù)池用Set結(jié)構(gòu)存儲，可以通過指針來刪除對應(yīng)的任務(wù)，通過閉包引用該指針從而達(dá)到動態(tài)控制并發(fā)池數(shù)目
for await 結(jié)構(gòu)體里，其實await下面，包括結(jié)構(gòu)體外都是屬于微任務(wù)（前提是有一個await里面的if被命中），至于這個微任務(wù)什么時候被加入微任務(wù)隊列，要看請求的那里的在什么時候開始標(biāo)記（resolve/reject ）
for await 里其實已經(jīng)在此輪宏任務(wù)當(dāng)中并發(fā)執(zhí)行了，await后面的代碼被掛起來，等前一個promise轉(zhuǎn)變狀態(tài)-->移出pool-->將下一個promise撈起加入pool當(dāng)中 -->下一個await等待最快的promise，如此往復(fù)。

可以想象這樣一個場景，幾組人在玩百米接力賽，每一組分別在0m,100m,200m的地方，有幾個賽道每組就有幾個人。（注意，這里想象成每個節(jié)點（比如0m處）這幾個人是一組），每到下一個節(jié)點的人，將棒子交給排隊在最前面的下一個人，下一個人就開始跑。

疑問

Promise.allSettled 和race 傳入的Promise<any>[]可以被其中的觸發(fā)微任務(wù)操作增減，這樣做會改變結(jié)果嗎？

有什么能拓展的功能呢

1.想要在執(zhí)行之后得到返回所需要的結(jié)果

（在第二種方法當(dāng)中已經(jīng)實現(xiàn)，第一種方法下可以通過增加一個 task->結(jié)果的map來收集，或者對所有的task分別包裹一層Promise，形成一個新的promiseList，放到Promise.allSettled里面，再把resolve以task->resolve的方式映射出來，在runner里面找到把Promise實例通過對應(yīng)的resolve暴露出去）

2.增加一個參數(shù)用來控制請求失敗的重試次數(shù)

拓展實現(xiàn)

增加重試次數(shù)以及回調(diào)函數(shù)增加返回結(jié)果

實現(xiàn)思路：

每一個請求額外包裹一層promise，形成一個新的promise數(shù)組，將此數(shù)組放入Promise.allSettled,回調(diào)函數(shù)在allSettled的then里面注冊。

將用來包裹的promise 里面的 resolve和reject以及剩余重試次數(shù)等信息包裝成對象，依次放入到用來執(zhí)行的隊列當(dāng)中。此隊列的作用為，執(zhí)行時取出，往后如果要重試，則重新加入到此隊列。

實現(xiàn)1的改造

增加參數(shù)retryTimes:number來表示重試次數(shù)

注意是重試次數(shù)，不是一共請求的次數(shù)。

大綱

function sendRequest(requestList, limits, callback, retryTimes) {
    // 定義執(zhí)行隊列，表示所有待執(zhí)行的任務(wù)
    const requestListWrapperedQueue = [];
    // 定義開始時能執(zhí)行的并發(fā)數(shù)
    const concurrentNum = Math.min(limits, requestList.length);
    // 定義放在allSettled的所有promise
    const returnPromises = []
    // 當(dāng)前并發(fā)數(shù)
    let concurrentCount = 0;
    // 新增： 包裹promise,并且將相關(guān)信息重新包裝放入請求隊列
    const wrapePromise = (requestItem)=>{}
    // 啟動初次能執(zhí)行的任務(wù)
    const runTaskNeeded = () => {}; // 取出任務(wù)并推送到執(zhí)行器
    const runTask = () => {};
    // 執(zhí)行器，這里去執(zhí)行任務(wù)
    const runner = (task) => {};
    // 撈起下一個任務(wù)
    const picker = () => {};
    // 新增： 初始化，構(gòu)建執(zhí)行隊列以及包裹promise
    const init = ()=>{}
    // 開始執(zhí)行函數(shù)
    const start = ()=>{}
    // 開始
    start()
    // 新增：
    Promise.allSettled(returnPromises).then(callback,callback)
}

完整實現(xiàn)

function sendRequest(requestList, limits, callback, retryTimes) {
    // 定義執(zhí)行隊列，表示所有待執(zhí)行的任務(wù)
    const requestListWrapperedQueue = [];
    // 定義開始時能執(zhí)行的并發(fā)數(shù)
    const concurrentNum = Math.min(limits, requestList.length);
    // 定義放在allSettled的所有promise
    const returnPromises = [];
    // 當(dāng)前并發(fā)數(shù)
    let concurrentCount = 0;
    // 新增： 包裹promise,并且將相關(guān)信息重新包裝放入請求隊列
    const wrapePromise = (requestItem)=>{
        return new Promise((resolve,reject)=>{
            // 構(gòu)建執(zhí)行隊列
            requestListWrapperedQueue.push({
                requestFn:requestItem,  // 請求函數(shù)放到此處
                resolve,
                reject,
                remainRetryTime:retryTimes // 剩余重試次數(shù)
            })
        })
    };
    // 啟動初次能執(zhí)行的任務(wù)
    const runTaskNeeded = () => {
        let i = 0
        // 啟動當(dāng)前的任務(wù)
        while(i < concurrentNum){
            i++
            runTask()
        }
    };
    // 取出任務(wù)并推送到執(zhí)行器
    const runTask = () => {
        const task = requestListWrapperedQueue.shift()
        task && runner(task)
    };
    // 執(zhí)行器，這里去執(zhí)行任務(wù)
    const runner = async (task) => {
        const {
            requestFn,
            resolve,
            reject,
            remainRetryTime
        } = task;
        try {
            // 并發(fā)數(shù) +1
            concurrentCount++
            // 執(zhí)行任務(wù)
            const res = await requestFn()
            // 拿到結(jié)果，直接結(jié)束
            resolve(res)
        } catch (error) {
            // 判斷還有無重試次數(shù)
            if(remainRetryTime > 0){
                // 重新放回隊列，注意這樣并不會影響allSettled結(jié)果的順序
                requestListWrapperedQueue.push(task)
                // 剩余重試次數(shù)-1
                task.remainRetryTime --
            }else {
                // 沒有剩余次數(shù)則直接結(jié)束
                reject(error)
            }
        }finally{
            // 并發(fā)數(shù)-1
            concurrentCount--
            // 撈起下一個任務(wù)
            picker()
        }
    };
    // 撈起下一個任務(wù)
    const picker = () => {
        if(concurrentCount < limits && requestListWrapperedQueue.length > 0 ){
            // 繼續(xù)執(zhí)行任務(wù)
            runTask()
        }
    };
    // 新增： 初始化，構(gòu)建執(zhí)行隊列以及包裹promise
    const init = ()=>{
        for(let requestItem of requestList){
            const wrapperedPromise = wrapePromise(requestItem)
            // 構(gòu)建包裹promise的數(shù)組，用于allSettled
            returnPromises.push(wrapperedPromise)
        }
    }
    // 開始執(zhí)行函數(shù)
    const start = ()=>{
        init()
        runTaskNeeded()
    }
    // 開始
    start()
    // 新增：allSettled用來獲取結(jié)果
    Promise.allSettled(returnPromises).then(callback,callback)
}