队列有x个之后执行
正文
在网上看到这么一道题：
首先来实现一个分割数组的函数~
const group = (list = [], max = 0) => {
if (!list.length) {
return list;
}
let results = [];
for (let i = 0, len = list.length; i < len; i += max) {
results.push(list.slice(i, i + max));
}
return results;
};
这里就是根据指定的并发数量来分割数组。主要就是for + slice，这没啥好说的
接下来再来一个用async + await实现的请求集合封装。
通Promise.allSettled去执行每一组的请求集合。
Promise.allSettled是一个新的 API，跟Promise.all差不多的用法，也是接受的数组，不过不同的是Promise.allSettled会等所有任务结束之后才会返回结果，而Promise.all只要有一个reject就会返回结果。
const requestHandler = async (groupedUrl = [], callback = () => {}) => {
if (!groupedUrl.length) {
callback();
return groupedUrl;
}
const newGroupedUrl = groupedUrl.map((fn) => fn());
const resultsMapper = (results) => results.map(callback);
const data = await Promise.allSettled(newGroupedUrl).then(resultsMapper);
return data;
};
接下来就是主函数
const sendRequest = async (urls = [], max = 0, callback = () => {}) => {
if (!urls.length) {
return urls;
}
const groupedUrls = group(urls, max);
const results = [];
console.log("start !");
for (let groupedUrl of groupedUrls) {
try {
const result = await requestHandler(groupedUrl, callback);
results.push(result);
console.log("go");
} catch {}
}
console.log("done !");
return results;
};
这里就是利用了for + async + await来限制并发。等每次并发任务结果出来之后再执行下一次的任务。
执行下栗子：
const p1 = () =>
new Promise((resolve, reject) => setTimeout(reject, 1000, "p1"));
const p2 = () => Promise.resolve(2);
const p3 = () =>
new Promise((resolve, reject) => setTimeout(resolve, 2000, "p3"));
const p4 = () => Promise.resolve(4);
const p5 = () =>
new Promise((resolve, reject) => setTimeout(reject, 2000, "p5"));
const p6 = () => Promise.resolve(6);
const p7 = () =>
new Promise((resolve, reject) => setTimeout(resolve, 1000, "p7"));
const p8 = () => Promise.resolve(8);
const p9 = () =>
new Promise((resolve, reject) => setTimeout(reject, 1000, "p9"));
const p10 = () => Promise.resolve(10);
const p11 = () =>
new Promise((resolve, reject) => setTimeout(resolve, 2000, "p10"));
const p12 = () => Promise.resolve(12);
const p13 = () =>
new Promise((resolve, reject) => setTimeout(reject, 1000, "p11"));
const p14 = () => Promise.resolve(14);
const ps = [p1, p2, p3, p4, p5, p6, p7, p8, p9, p10, p11, p12, p13, p14];
sendRequest(ps, 3, ({ reason, value }) => {
console.log(reason || value);
});
队列最多有x个
前言：
- 浏览器对同一域名下同一时间点的最大连接数做了限制，谷歌是 6 个，其他浏览器可百度查看相关资料。
- 浏览器同一时间点内发送的请求过多，会导致请求很慢页面卡顿的情况
解决问题
- 封装限制 Promise 异步任务并发请求数
核心函数也就两个。
调用器：就是把真正的执行函数和参数传入，创建返回一个新的 Promise，而这个新 Promise 的什么时候返回，取决于这个异步任务何时被调度。Promise 内部主要就是创建一个任务，判断任务是执行还是入队。
创建任务：实际上就是返回了一个函数，将真正的执行函数放在里面执行。这里利用了 Promise 的 finally 方法，在 finally 中判断是否执行下一个任务，实现任务队列连续消费的地方就是这里。
/**
* 封装axios并发请求数
*/
class LimitPromise {
constructor(max) {
// 异步任务“并发”上限
this._max = max || 6;
// 当前正在执行的任务数量
this._count = 0;
// 等待执行的任务队列
this._taskQueue = [];
}
/**
* 调用器，将异步任务函数和它的参数传入
* @param caller 异步任务函数，它必须是async函数或者返回Promise的函数
* @param args 异步任务函数的参数列表
* @returns {Promise<unknown>} 返回一个新的Promise
*/
call(caller, ...args) {
return new Promise((resolve, reject) => {
const task = this._createTask(caller, args, resolve, reject);
if (this._count >= this._max) {
// console.log('count >= max, push a task to queue', this._count , this._max, this._taskQueue)
this._taskQueue.push(task);
} else {
// console.log('数组中的对列长度还没超过6个', this._count)
task();
}
});
}
/**
* 创建一个任务
* @param caller 实际执行的函数
* @param args 执行函数的参数
* @param resolve
* @param reject
* @returns {Function} 返回一个任务函数
* @private
*/
_createTask(caller, args, resolve, reject) {
return () => {
// 实际上是在这里调用了异步任务，并将异步任务的返回（resolve和reject）抛给了上层
caller(...args)
.then(resolve)
.catch(reject)
.finally(() => {
// 任务队列的消费区，利用Promise的finally方法，在异步任务结束后，取出下一个任务执行
this._count--;
if (this._taskQueue.length) {
// console.log('a task run over, pop a task to run', this._taskQueue)
let task = this._taskQueue.shift();
task();
} else {
// console.log('task count = ', count)
}
});
this._count++;
// console.log('task run , task count = ', this._count)
};
}
}
export default LimitPromise;
- 在封装的 axios 文件使用并发数限制
import axios from "axios";
import { message } from "antd";
import qs from "qs";
// 引入上面封装好的并发数限制文件
import LimitPromise from "./limitPromise";
// 并发请求上限
const MAX = 5;
// 调用核心控制器类
const limitP = new LimitPromise(MAX);
//响应时间
axios.defaults.timeout = 300000; // 5 min
axios.defaults.headers.post["Content-Type"] = "application/json";
// request interceptor
axios.interceptors.request.use(
(config) => {
config.headers["Content-Type"] = "application/json";
config.headers["datae-token"] = localStorage.getItem("xyToken");
//application/x-www-form-urlencoded
if (config.resType === "form") {
config.headers["Content-Type"] = "application/x-www-form-urlencoded";
config.data = qs.stringify(config.data);
return config;
}
return config;
},
(error) => {
return Promise.reject(error);
}
);
// response interceptor
axios.interceptors.response.use(
(response) => {
if (response.status >= 400) {
// 没登录/过期
if (response.status === 401) {
message.error("登录态过期，请重新登录");
// ...写入自己的代码逻辑和判断
return;
}
return Promise.reject(response);
}
return response;
},
(error) => {
if (error.response.status === 401) {
message.error("登录态过期，请重新登录");
// ...写入自己的代码逻辑和判断
} else {
message.error("网络连接超时");
}
return Promise.reject(error);
}
);
// 导出axios的几种请求方式
export const getP = (url, params, config = {}) => {
return limitP
.call(axios.get, url, {
params: params,
...config,
})
.then((res) => res.data);
};
export const postP = (url, params, config = {}) => {
return limitP
.call(axios.post, url, params, {
...config,
})
.then((res) => res.data);
};
// 如果还需要其他请求，请结合aixos的api进行相关配置
export default axios;