JS使用Promise控制请求并发数：实现思路与代码详解

前言

在前端面试中，手写代码题是不可或缺的环节，其中请求并发控制是高频考点。由于现代前端项目普遍使用axios封装异步请求，这道题实际上是在考察开发者对Promise机制及异步编程原理的掌握程度。

引出

题目：

// 设计一个函数，可以限制请求的并发，同时请求结束之后，调用callback函数
// sendRequest(requestList:,limits,callback):void
sendRequest(
[()=>request('1'),
()=>request('2'),
()=>request('3'),
()=>request('4')],
3, //并发数
(res)=>{
    console.log(res)
})
// 其中request 可以是： 
function request (url,time=1){
    return new Promise((resolve,reject)=>{
        setTimeout(()=>{
            console.log('请求结束：'+url);
            if(Math.random() > 0.5){
                resolve('成功')
            }else{
                reject('错误;')
            }
        },time*1e3)
    })
}

明确概念

在开始编码之前，需要厘清几个核心概念：

并发（Concurrency）：指多个任务在时间片上交替执行。由于CPU处理指令的速度极快，它无需按顺序逐段执行代码，这种交替执行在宏观上表现为“同时”进行，故称为并发。
并行（Parallelism）：指在多个物理CPU核心或分布式系统中，任务真正意义上地“同时”执行。
并发控制：指管理多个并发任务，一旦有任务完成，立即启动下一个等待中的任务，以维持设定的并发数量。
切片控制：将并发任务划分为若干批次（例如10个任务分为2批，每批5个）。只有当前批次的所有任务执行完毕后，才开始执行下一批。这种方式虽然也能控制并发，但效率通常低于动态的并发控制策略。

思路

基本思路是：首先启动允许执行的并发任务。根据并发的定义，每当一个任务执行完毕，就立即从队列中取出下一个任务加入执行。

为了代码的可读性和可维护性，应将关键步骤拆分为独立的函数来组织逻辑。

循环启动可执行的任务。
从任务列表中取出任务，并将其推入执行器。
在执行器内部，更新当前的并发计数，并在任务完成后触发“捞起”逻辑以执行下一个任务。
在“捞起”任务的逻辑中，可以触发最终的回调函数，并继续调用执行器以执行剩余任务。

实现

1.定义常量和函数

function sendRequest(requestList,limits,callback){
    // 定义执行队列，表示所有待执行的任务
    const promises = requestList.slice()
    // 定义开始时能执行的并发数
    const concurrentNum = Math.min(limits,requestList.length)
    let concurrentCount = 0 // 当前并发数
    // 启动初次能执行的任务
    const runTaskNeeded = ()=>{
        let i = 0
        while(i<concurrentNum){
            runTask()
        }
    }
    // 取出任务并推送到执行器
    const runTask = ()=>{}
    // 执行器，这里去执行任务
    const runner = ()=>{}
    // 捞起下一个任务
    const picker = ()=>{}
    // 开始执行！
    runTaskNeeded()
}

2.实现对应的函数

function sendRequest(requestList,limits,callback){
    const promises = requestList.slice() // 取得请求list（浅拷贝一份）
    // 得到开始时，能执行的并发数
    const concurrentNum = Math.min(limits,requestList.length)
    let concurrentCount = 0 // 当前并发数
    // 第一次先跑起可以并发的任务
    const runTaskNeeded = ()=>{
        let i = 0
        // 启动当前能执行的任务
        while(i<concurrentNum){
            i++
            runTask()
        }
    }
    // 取出任务并且执行任务
    const runTask = ()=>{
        const task = promises.shift()
        task && runner(task)
    }
    // 执行器
    // 执行任务，同时更新当前并发数
    const runner = async (task)=>{
        try {
            concurrentCount++
            await task()
        } catch (error) {
        }finally{
            // 并发数--
            concurrentCount--
            // 捞起下一个任务
            picker()
        }
    }
// 捞起下一个任务
    const picker = ()=>{
        // 任务队列里还有任务并且此时还有剩余并发数的时候 执行
        if(concurrentCount < limits && promises.length > 0 ){
            // 继续执行任务
            runTask()
        // 队列为空的时候，并且请求池清空了，就可以执行最后的回调函数了
        }else if(promises.length ==0 && concurrentCount ==0 ){
            // 执行结束
            callback && callback()
        }
    }
    // 入口执行
    runTaskNeeded()
}

另一种实现

核心逻辑在于判断：当【有任务执行完成】时，再检查是否还有剩余任务可执行。可以维护一个代表当前正在执行任务的池（pool），利用`await Promise.race(pool)`来等待池中最快的任务完成，从而判断是否有任务结束。

async function sendRequest(requestList,limits,callback){
    // 维护一个promise队列
    const promises = []
    // 当前的并发池,用Set结构方便删除
    const pool = new Set() // set也是Iterable<any>[]类型，因此可以放入到race里
    // 开始并发执行所有的任务
    for(let request of requestList){
        // 开始执行前，先await 判断 当前的并发任务是否超过限制
        if(pool.size >= limits){
            // 这里因为没有try catch ，所以要捕获一下错误，不然影响下面微任务的执行
            await Promise.race(pool)
            .catch(err=>err)
        }
        const promise = request()// 拿到promise
        // 删除请求结束后，从pool里面移除
        const cb = ()=>{
            pool.delete(promise)
        }
        // 注册下then的任务
        promise.then(cb,cb)
        pool.add(promise)
        promises.push(promise)
    }
    // 等最后一个for await 结束，这里是属于最后一个 await 后面的 微任务
    // 注意这里其实是在微任务当中了，当前的promises里面是能确保所有的promise都在其中(前提是await那里命中了if)
    Promise.allSettled(promises).then(callback,callback)
}

总结一下要点：

利用`race`的特性可以找到并发任务中最快结束的那个请求。
利用`for await`结构可以保证`for`循环体下方的代码是最后一个`await`后的微任务。而在最后一个微任务中，可以确保所有的Promise都已存入`promises`数组（即使未命中任何`await`，即限制并发数大于任务数时，也能保证所有Promise都在其中）。最后利用`allSettled`等待所有Promise状态转变后，调用回调函数。
并发任务池使用Set结构存储，可以通过指针删除对应的任务，并通过闭包引用该指针，从而动态控制并发池的大小。
在`for await`结构中，其实`await`语句下方以及结构体外都属于微任务（前提是有一个`await`内部的`if`被命中）。至于这个微任务何时被加入微任务队列，取决于请求在何时被标记（resolve/reject）。
在`for await`循环内部，实际上已经在此轮宏任务中并发执行了。`await`后面的代码被挂起，等待前一个Promise状态转变后移出pool，再将下一个Promise捞起加入pool，随后下一个`await`等待最快的Promise，如此往复。

可以想象这样一个场景：几组人在玩百米接力赛，每一组分别在0m, 100m, 200m的地方，有几个赛道每组就有几个人。（注意，这里想象成每个节点（比如0m处）这几个人是一组），每到下一个节点的人，将棒子交给排队在最前面的下一个人，下一个人就开始跑。

疑问

Promise.allSettled 和race 传入的Promise<any>[]可以被其中的触发微任务操作增减，这样做会改变结果吗？

有什么能拓展的功能呢

1.想要在执行之后得到返回所需要的结果

（在第二种方法当中已经实现，第一种方法下可以通过增加一个 task->结果的map来收集，或者对所有的task分别包裹一层Promise，形成一个新的promiseList，放到Promise.allSettled里面，再把resolve以task->resolve的方式映射出来，在runner里面找到把Promise实例通过对应的resolve暴露出去）

2.增加一个参数用来控制请求失败的重试次数

拓展实现

增加重试次数以及回调函数增加返回结果

实现思路：

每一个请求额外包裹一层promise，形成一个新的promise数组，将此数组放入Promise.allSettled,回调函数在allSettled的then里面注册。

将用来包裹的promise 里面的 resolve和reject以及剩余重试次数等信息包装成对象，依次放入到用来执行的队列当中。此队列的作用为，执行时取出，往后如果要重试，则重新加入到此队列。

实现1的改造

增加参数retryTimes:number来表示重试次数

注意是重试次数，不是一共请求的次数。

大纲

function sendRequest(requestList, limits, callback, retryTimes) {
    // 定义执行队列，表示所有待执行的任务
    const requestListWrapperedQueue = [];
    // 定义开始时能执行的并发数
    const concurrentNum = Math.min(limits, requestList.length);
    // 定义放在allSettled的所有promise
    const returnPromises = []
    // 当前并发数
    let concurrentCount = 0;
    // 新增： 包裹promise,并且将相关信息重新包装放入请求队列
    const wrapePromise = (requestItem)=>{}
    // 启动初次能执行的任务
    const runTaskNeeded = () => {}; // 取出任务并推送到执行器
    const runTask = () => {};
    // 执行器，这里去执行任务
    const runner = (task) => {};
    // 捞起下一个任务
    const picker = () => {};
    // 新增： 初始化，构建执行队列以及包裹promise
    const init = ()=>{}
    // 开始执行函数
    const start = ()=>{}
    // 开始
    start()
    // 新增：
    Promise.allSettled(returnPromises).then(callback,callback)
}

完整实现

function sendRequest(requestList, limits, callback, retryTimes) {
    // 定义执行队列，表示所有待执行的任务
    const requestListWrapperedQueue = [];
    // 定义开始时能执行的并发数
    const concurrentNum = Math.min(limits, requestList.length);
    // 定义放在allSettled的所有promise
    const returnPromises = [];
    // 当前并发数
    let concurrentCount = 0;
    // 新增： 包裹promise,并且将相关信息重新包装放入请求队列
    const wrapePromise = (requestItem)=>{
        return new Promise((resolve,reject)=>{
            // 构建执行队列
            requestListWrapperedQueue.push({
                requestFn:requestItem,  // 请求函数放到此处
                resolve,
                reject,
                remainRetryTime:retryTimes // 剩余重试次数
            })
        })
    };
    // 启动初次能执行的任务
    const runTaskNeeded = () => {
        let i = 0
        // 启动当前的任务
        while(i < concurrentNum){
            i++
            runTask()
        }
    };
    // 取出任务并推送到执行器
    const runTask = () => {
        const task = requestListWrapperedQueue.shift()
        task && runner(task)
    };
    // 执行器，这里去执行任务
    const runner = async (task) => {
        const {
            requestFn,
            resolve,
            reject,
            remainRetryTime
        } = task;
        try {
            // 并发数 +1
            concurrentCount++
            // 执行任务
            const res = await requestFn()
            // 拿到结果，直接结束
            resolve(res)
        } catch (error) {
            // 判断还有无重试次数
            if(remainRetryTime > 0){
                // 重新放回队列，注意这样并不会影响allSettled结果的顺序
                requestListWrapperedQueue.push(task)
                // 剩余重试次数-1
                task.remainRetryTime --
            }else {
                // 没有剩余次数则直接结束
                reject(error)
            }
        }finally{
            // 并发数-1
            concurrentCount--
            // 捞起下一个任务
            picker()
        }
    };
    // 捞起下一个任务
    const picker = () => {
        if(concurrentCount < limits && requestListWrapperedQueue.length > 0 ){
            // 继续执行任务
            runTask()
        }
    };
    // 新增： 初始化，构建执行队列以及包裹promise
    const init = ()=>{
        for(let requestItem of requestList){
            const wrapperedPromise = wrapePromise(requestItem)
            // 构建包裹promise的数组，用于allSettled
            returnPromises.push(wrapperedPromise)
        }
    }
    // 开始执行函数
    const start = ()=>{
        init()
        runTaskNeeded()
    }
    // 开始
    start()
    // 新增：allSettled用来获取结果
    Promise.allSettled(returnPromises).then(callback,callback)
}

结尾

这种题目是考验你对异步编程的理解，要想写出来，你需要具备事件循环以及promise的知识。

到此这篇关于JS使用Promise控制请求并发数的文章就介绍到这了,更多相关JS控制请求并发数内容请搜索本站以前的文章或继续浏览下面的相关文章希望大家以后多多支持本站！

JS使用Promise控制请求并发数

前言

引出

明确概念

思路

实现

另一种实现

有什么能拓展的功能呢

拓展实现

结尾

实用工具推荐