本文由 简悦 SimpRead 转码， 原文地址 mp.weixin.qq.com
写在前面
最近在进行移动端 h5 开发，首页需要加载的资源很多，一个 lottie 动效需要请求 70 多张图片，但是遇到安卓 webview 限制请求并发数，导致部分图片请求失败破图。当然图片资源可以做闲时加载和预加载，可以减轻播放动效时资源未加载的问题。
同样的，业务开发也会遇到需要异步请求几十个接口，如果同时并发请求浏览器会进行限制请求数，也会给后端造成请求压力。
场景说明
现在有个场景：
请你实现一个并发请求函数 concurrencyRequest(urls, maxNum)，要求如下：
要求最大并发数 maxNum
每当有一个请求返回，就留下一个空位，可以增加新的请求
所有请求完成后，结果按照 urls 里面的顺序依次打出（发送请求的函数可以直接使用 fetch 即可）
初始实现：
const preloadManger = (urls, maxCount = 5) => { let count = 0; // 计数 -- 用于控制并发数 const createTask = () => { if (count < maxCount) { const url = urls.pop(); // 从请求数组中取值 if (url) { // 无论请求是否成功，都要执行taskFinish loader(url).finally(taskFinish); // 添加下一个请求 count++; createTask(); } } }; const taskFinish = () => { count--; createTask(); }; createTask();};// 进行异步请求const loader = async (url) => { const res = await fetch(url).then(res=>res.json()); console.log("res",res); return res}const urls = [];for (let i = 1; i <= 20; i++) { urls.push(`https://jsonplaceholder.typicode.com/todos/${i}`);}preloadManger(urls, 5)请求状态：
可以看到上面的请求是每五个一组进行请求，当一个请求无论返回成功或是失败，都会从请求数组中再取一个请求进行补充。
设计思路
那么，我们可以考虑使用队列去请求大量接口。
思路如下：
假定最大并发数是 maxNum=5，图中对接口进行了定义编号，当请求队列池中有一个请求返回后，就向池子中新增一个接口进行请求，依次直到最后一个请求执行完毕。
当然，要保证程序的健壮性，需要考虑一些边界情况，如下：
当初始请求数组 urls 的长度为 0 时，此时请求结果数组 results 是个空数组
最大并发数 maxNums>urls 的长度时，请求数为 urls 的长度
需要定义计数器 count 去判断是否全部请求完毕
无论请求成功与否，都应该将结果存在结果数组 results 中
结果数组 results 和 urls 数组的顺序保持一致，方便存取
代码实现
在前面的初始实现的代码中，虽然都能满足基本需求，但是并没有考虑一些边界条件，对此需要根据上面设计思路重新实现得到：
// 并发请求函数const concurrencyRequest = (urls, maxNum) => { return new Promise((resolve) => { if (urls.length === 0) { resolve([]); return; } const results = []; let index = 0; // 下一个请求的下标 let count = 0; // 当前请求完成的数量 // 发送请求 async function request() { if (index === urls.length) return; const i = index; // 保存序号，使result和urls相对应 const url = urls[index]; index++; console.log(url); try { const resp = await fetch(url); // resp 加入到results results[i] = resp; } catch (err) { // err 加入到results results[i] = err; } finally { count++; // 判断是否所有的请求都已完成 if (count === urls.length) { console.log('完成了'); resolve(results); } request(); } } // maxNum和urls.length取最小进行调用 const times = Math.min(maxNum, urls.length); for(let i = 0; i < times; i++) { request(); } })}测试代码：
const urls = [];for (let i = 1; i <= 20; i++) { urls.push(`https://jsonplaceholder.typicode.com/todos/${i}`);}concurrencyRequest(urls, 5).then(res => { console.log(res);})请求结果：
上面代码基本实现了前端并发请求的需求，也基本满足需求，在生产中其实有很多已经封装好的库可以直接使用。比如：p-limit【https://github.com/sindresorhus/p-limit】
阅读 p-limit 源码
import Queue from 'yocto-queue';import {AsyncResource} from '#async_hooks';export default function pLimit(concurrency) { // 判断这个参数是否是一个大于0的整数，如果不是就抛出一个错误 if ( !((Number.isInteger(concurrency) || concurrency === Number.POSITIVE_INFINITY) && concurrency > 0) ) { throw new TypeError('Expected `concurrency` to be a number from 1 and up'); } // 创建队列 -- 用于存取请求 const queue = new Queue(); // 计数 let activeCount = 0; // 用来处理并发数的函数 const next = () => { activeCount--; if (queue.size > 0) { // queue.dequeue()可以理解为[].shift()，取出队列中的第一个任务，由于确定里面是一个函数，所以直接执行就可以了； queue.dequeue()(); } }; // run函数就是用来执行异步并发任务 const run = async (function_, resolve, arguments_) => { // activeCount加1，表示当前并发数加1 activeCount++; // 执行传入的异步函数，将结果赋值给result，注意：现在的result是一个处在pending状态的Promise const result = (async () => function_(...arguments_))(); // resolve函数就是enqueue函数中返回的Promise的resolve函数 resolve(result); // 等待result的状态发生改变，这里使用了try...catch，因为result可能会出现异常，所以需要捕获异常； try { await result; } catch {} next(); }; // 将run函数添加到请求队列中 const enqueue = (function_, resolve, arguments_) => { queue.enqueue( // 将run函数绑定到AsyncResource上，不需要立即执行，对此添加了一个bind方法 AsyncResource.bind(run.bind(undefined, function_, resolve, arguments_)), ); // 立即执行一个异步函数，等待下一个微任务（注意：因为activeCount是异步更新的，所以需要等待下一个微任务执行才能获取新的值） (async () => { // This function needs to wait until the next microtask before comparing // `activeCount` to `concurrency`, because `activeCount` is updated asynchronously // when the run function is dequeued and called. The comparison in the if-statement // needs to happen asynchronously as well to get an up-to-date value for `activeCount`. await Promise.resolve(); // 判断activeCount是否小于concurrency，并且队列中有任务，如果满足条件就会将队列中的任务取出来执行 if (activeCount < concurrency && queue.size > 0) { // 注意：queue.dequeue()()执行的是run函数 queue.dequeue()(); } })(); }; // 接收一个函数fn和参数args，然后返回一个Promise，执行出队操作 const generator = (function_, ...arguments_) => new Promise(resolve => { enqueue(function_, resolve, arguments_); }); // 向外暴露当前的并发数和队列中的任务数，并且手动清空队列 Object.defineProperties(generator, { // 当前并发数 activeCount: { get: () => activeCount, }, // 队列中的任务数 pendingCount: { get: () => queue.size, }, // 清空队列 clearQueue: { value() { queue.clear(); }, }, }); return generator;}整个库只有短短 71 行代码，在代码中导入了 yocto-queue 库，它是一个微型的队列数据结构。
手写源码
在进行手撕源码时，可以借助数组进行简易的实现：
class PLimit { constructor(concurrency) { this.concurrency = concurrency; this.activeCount = 0; this.queue = []; return (fn, ...args) => { return new Promise(resolve => { this.enqueue(fn, resolve, args); }); } } enqueue(fn, resolve, args) { this.queue.push(this.run.bind(this, fn, resolve, args)); (async () => { await Promise.resolve(); if (this.activeCount < this.concurrency && this.queue.length > 0) { this.queue.shift()(); } })(); } async run(fn, resolve, args) { this.activeCount++; const result = (async () => fn(...args))(); resolve(result); try { await result; } catch { } this.next(); } next() { this.activeCount--; if (this.queue.length > 0) { this.queue.shift()(); } }}小结
在这篇文章中，简要介绍了为什么要进行并发请求，阐述了使用请求池队列实现并发请求的设计思路，简要实现代码。
此外，还阅读分析了 p-limit 的源码，并使用数组进行简要的源码编写，以实现要求。
参考文章
【源码共读】大并发量如何控制并发数 https://juejin.cn/post/7179220832575717435?searchId=20240430092814392DC2208C545E691A26
前端实现并发控制网络请求 https://mp.weixin.qq.com/s/9uq2SqkcMSSWjks0x7RQJg
关于前端：如何实现并发请求数量控制？https://juejin.cn/post/7163522138698153997