控制请求并发数量:p-limit 源码解读

news/2024/11/18 2:47:26/文章来源:https://www.cnblogs.com/zsxblog/p/18426066

p-limit 是一个控制请求并发数量的库,他的整体代码不多,思路挺好的,很有学习价值;

举例

当我们同时发起多个请求时,一般是这样做的

Promise.all([requestFn1,requestFn2,requestFn3
]).then(res =>{})

或者

requestFn1()
requestFn2()
requestFn3()

而使用 p-limit 限制并发请求数量是这样做的:

var limit = pLimit(8); // 设置最大并发数量为 8var input = [ // Limit函数包装各个请求limit(() => fetchSomething('1')),limit(() => fetchSomething('2')),limit(() => fetchSomething('3')),limit(() => fetchSomething('4')),limit(() => fetchSomething('5')),limit(() => fetchSomething('6')),limit(() => fetchSomething('7')),limit(() => fetchSomething('8')),
];// 执行请求
Promise.all(input).then(res =>{console.log(res)
})

上面 input 数组包含了 8limit 函数,每个 limit 函数包含了要发起的请求

当设置最大并发数量为 8 时,上面 8 个请求会同时执行

来看下效果,假设每个请求执行时间为1s

var fetchSomething = (str) => {return new Promise((resolve, reject) => {setTimeout(() => {console.log(str)resolve(str)}, 1000)})
}

当设置并发请求数量为 2

image

当设置并发请求数量为 3

image

p-limit 限制并发请求数量本质上是,在内部维护了一个请求队列;

当请求发起时,先将请求推入队列,判断当前执行的请求数量是否小于配置的请求并发数量,如果是则执行当前请求,否则等待正在发起的请求中谁请求完了,再从队列首部取出一个执行;

源码(v2.3.0)

pLimit 源码如下(这个源码是 v2.3.0 版本的,因为项目中引入的版本比较早。后面会分析从 2.3.0 到最新版本的源码,看看增加或者改进了什么):

'use strict';
const pTry = require('p-try');const pLimit = concurrency => {// 限制为正整数if (!((Number.isInteger(concurrency) || concurrency === Infinity) && concurrency > 0)) {return Promise.reject(new TypeError('Expected `concurrency` to be a number from 1 and up'));}const queue = []; // 请求队列let activeCount = 0; // 当前并发的数量const next = () => { // 一个请求完成时执行的回调activeCount--;if (queue.length > 0) {queue.shift()();}};const run = (fn, resolve, ...args) => { // 请求开始执行activeCount++;const result = pTry(fn, ...args);resolve(result); // 将结果传递给 generatorresult.then(next, next); // 请求执行完调用回调};// 将请求加入队列const enqueue = (fn, resolve, ...args) => {if (activeCount < concurrency) {run(fn, resolve, ...args);} else {queue.push(run.bind(null, fn, resolve, ...args));}};const generator = (fn, ...args) => new Promise(resolve => enqueue(fn, resolve, ...args));// 暴露内部属性给外界Object.defineProperties(generator, {activeCount: {get: () => activeCount},pendingCount: {get: () => queue.length},clearQueue: {value: () => {queue.length = 0;}}});return generator;
};module.exports = pLimit;
module.exports.default = pLimit;

下面一一剖析下

1、pLimit 函数整体是一个闭包函数,返回了一个名叫 generator 的函数,由 generator 处理并发逻辑,
generator 返回值必须是 promise,这样才能被 Promise.all 捕获到

const generator = (fn,...args) => new Promise((resolve,reject)=7enqueue(fn,resolve,...args))

2、在 enqueue 函数里面

// 将请求加入队列
const enqueue = (fn, resolve, ...args) => {if (activeCount < concurrency) {run(fn, resolve, ...args);} else {queue.push(run.bind(null, fn, resolve, ...args));}
};

activeCount 表示正在执行的请求数量,当 activeCount 小于配置的并发数量(concurrency)时,则可以执行当前的 fn(执行 run 函数),否则推入请求队列等待。

3、run 函数接收了三个形参

const run = (fn, resolve, ...args) => { // 请求开始执行activeCount++;const result = pTry(fn, ...args);resolve(result);result.then(next, next);
};
  • fn 表示执行的请求,

  • resolvegenerator 定义并往下传,一直跟踪到请求执行完毕后,调用 resolve(result); 代表 generator 函数 fulfilled

  • ···args 表示其余的参数,最终会作为 fn 的参数。

4、执行 run 函数时

const run = (fn, resolve, ...args) => { // 请求开始执行activeCount++; // 请求开始执行,当前请求数量 +1const result = pTry(fn, ...args);resolve(result);result.then(next, next);
};

这里执行 fn 使用的是 const result = pTry(fn,...args)pTry 的作用就是创建一个 promise 包裹的结果,不论 fn 是同步函数还是异步函数

// pTry 源码
const pTry = (fn,...args) => new Promise((resolve,reject) => resolve(fn(...args)));

现在 fn 执行(fn(...args))完毕并兑现(resolve(fn(...args)))之后,result 就会兑现。

result 兑现后,generatorpromise 也就兑现了( resolve(result) ),那么当前请求 fn 的流程就执行完了。

5、当前请求执行完后,对应的当前正在请求的数量也要减一,activeCount--

const next = () => { // 一个请求完成时执行的回调activeCount--;if (queue.length > 0) {queue.shift()();}
};

然后继续从队列头部取出请求来执行

6、最后暴露内部属性给外界

Object.defineProperties(generator, {activeCount: { // 当前正在请求的数量get: () => activeCount},pendingCount: { // 等待执行的数量get: () => queue.length},clearQueue: {value: () => {queue.length = 0;}}
});

源码(v2.3.0)=> 源码(v6.1.0)

从 v2.3.0 到最新的 v6.1.0 版本中间加了一些改进

1、v3.0.0:始终异步执行传进 limit 的函数

image

3.0.0 中,作者将请求入队放在前面,将 if 判断语句和请求执行置于微任务中运行;正如源码注释中解释的:因为当 run 函数执行时,activeCount 是异步更新的,那么这里的 if 判断语句也应该异步执行才能实时获取到 activeCount 的值。

这样一开始批量执行 limit(fn) 时,将会先把这些请求全部放入队列中,然后再根据条件判断是否执行请求;

2、v3.0.2:修复传入的无效并发数引起的错误;

image

return Promise.reject 改为了直接 throw 一个错误

3、v3.1.0:移除 pTry 的依赖;改善性能;

image

移除了 pTry 依赖,改为了 async 包裹,上面有提到,pTry 是一个 promise 包装函数,返回结果是一个 promise;两者本质都是一样;

增加了 yocto-queue 依赖,yocto-queue是一个队列数据结构,用队列代替数组,性能更好;队列的入队和出队操作时间复杂度是 O(1),而数组的 shift()O(n);

4、v5.0.0:修复上下文传播问题

image

引入了 AsyncResource

export const AsyncResource = {bind(fn, _type, thisArg) {return fn.bind(thisArg);}
}

这里用 AsyncResource.bind() 包裹 run.bind(undefined, fn, resolve, args) ,其实不是太明白为啥加这一层。。。这里用的到三个参数(fn,resolve,args)都是通过函数传参过来的,和 this 没关系吧,各位知道的可以告知下么。

相关 issue 在这里

5、6.0.0:性能优化,主要优化的地方在下面

image

移除了 AsyncResource.bind(),改为使用一个立即执行的 promise,并将 promiseresolve 方法插入队列,一旦 resolve 完成兑现,调用相应请求;相关 issue 在这里

6、v6.1.0:允许实时修改并发限制数

image

改变并发数后立马再检测是否可以执行请求;


最后

在上面第4点的,第5点中的优化没太看明白,因为执行请求用的到三个参数(fn,resolve,args)都是通过函数传参过来的,看起来 this 没关系,为啥要进行多层 bind 绑定呢?各位知道的可以不吝赐教下么。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.hqwc.cn/news/801758.html

如若内容造成侵权/违法违规/事实不符,请联系编程知识网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

程序员职业发展之路思考:工程师的等级阶梯

德雷福斯模型:新手到专家 德雷福斯模型(Dreyfus model)是在 1980 年,Dreyfus 兄弟共同提出的技能习得模型。 它是一个技能习得的阶梯模型,也可以用来考察行业技术能手的分级。该模型由上而下分成:专家、精通者、胜任者、高级新手、新手五个等级,越到上面人数占比越少。新…

2024 人工智能学习内容

第六组思维导图:图形的认识

04. 流程控制

一、流程控制流程控制就是用来控制程序运行中各语句执行顺序的语句。基本的流程结构为:顺序结构,分支结构(或称选择结构),循环结构。顺序结构:程序自上到下执行,中间没有任何判断和跳转; 分支结构:根据条件,选择性的执行某段代码,有 if……else 和 switch……case 两…

CentOS 7 虚拟机连接网络

CentOS 7 虚拟机连接网络 检查网络 ping www.baidu.com切换 root 用户 su查看网卡名 ip addr激活网卡 vim /etc/sysconfig/network-scripts/ifcfg-ens33重启网络 service network restart

execve

目录glibc glibc execve() 执行由 pathname 指定的程序。这会导致当前正在被调用进程运行的程序被一个新程序替换,且该新程序会重新初始化栈、堆,以及(已初始化和未初始化的)数据段。

freeRTOS源码解析4--tasks.c 5

4.2.13 继续任务--vTaskResume 接口:void vTaskResume( TaskHandle_t xTaskToResume )形参1:xTaskToResume ,想要继续的任务handle; 首先是vTaskResume调用的一个内部函数:static BaseType_t prvTaskIsTaskSuspended( const TaskHandle_t xTask ),用于检查任务是否是挂起…

MySQL 必知概念

Delete、Drop 和 Truncatedelete、truncate 仅仅删除表里面的数据,drop会把表的结构也删除 delete 是 DML 语句,操作完成后,可以回滚,truncate 和 drop 是 DDL 语句,删除之后立即生效,不能回滚 执行效率:drop > truncate > deleteMyISAM 与 InnoDBInnoDB 支持事务…

视野修炼-技术周刊第102期 | js 编译运行C

① Bun 现在允许直接在js中直接编译运行 C ! ② caniuse-cli ③ SSL证书管理工具 ④ 好的重构与坏的重构 ⑤ sisi - 命令行图片检索工具 ⑥ cvbee.ai - AI 简历生成欢迎来到第 102 期的【视野修炼 - 技术周刊】,下面是本期的精选内容简介 🔥强烈推荐Bun 现在允许直接在js中…

【vulhub】Discuz-命令执行 wooyun-2010-080723

【vulhub】Discuz-命令执行 wooyun-2010-080723 ​docker-compose up-d​启动! ​​ wooyun-2010-080723 命令执行 0x01 搭建环境 访问192.168.132.138:8080/install​,安装数据库。数据库服务器填写db(必须db,不然安装失败),数据库名为discuz,数据库账号密码均为root,…

华科python与人工智能实践(公选)教程

python基础 软件下载 1.python下载安装 点击此链接进入官网windows下载地址点击箭头处链接下载最新版本,进入页面后下拉根据你的机器下载对应版本,一般人使用的是X86架构windos系统,下载箭头所指即可 若是不知道CPU架构,可见查看cpu架构,x86还是arm 下载后根据指引进行安装…

2376.统计特殊整数

如果一个正整数每一个数位都是 互不相同 的,我们称它是 特殊整数 。 给你一个 正 整数 n ,请你返回区间 [1, n] 之间特殊整数的数目。 示例 1: 输入:n = 20 输出:19 解释:1 到 20 之间所有整数除了 11 以外都是特殊整数。所以总共有 19 个特殊整数。 示例 2: 输入:n = …

数业智能心大陆:职场倦怠的新解法

什么是职业倦怠? 在职场中,职业倦怠的表现形式丰富多样。从数业智能心大陆 AI 心理咨询平台的数据来看,职业倦怠呈现出多种状态。教师可能对教学不再满怀热情,精心备课也成为过去式;情绪上容易烦躁、易怒,在工作压力之下,常常因为一些小事就被激怒。比如在项目团队中,成…