- VisualStudio2022插件的安装及使用-编程手把手系列文章
- pprof-在现网场景怎么用
- C#实现的下拉多选框,下拉多选树,多级节点
- 【学习笔记】基础数据结构:猫树
p-limit 是一个控制请求并发数量的库,他的整体代码不多,思路挺好的,很有学习价值; 。
当我们同时发起多个请求时,一般是这样做的 。
Promise.all([
requestFn1,
requestFn2,
requestFn3
]).then(res =>{})
或者 。
requestFn1()
requestFn2()
requestFn3()
而使用 p-limit 限制并发请求数量是这样做的:
var limit = pLimit(8); // 设置最大并发数量为 8
var 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 数组包含了 8 个 limit 函数,每个 limit 函数包含了要发起的请求 。
当设置最大并发数量为 8 时,上面 8 个请求会同时执行 。
来看下效果,假设每个请求执行时间为1s.
var fetchSomething = (str) => {
return new Promise((resolve, reject) => {
setTimeout(() => {
console.log(str)
resolve(str)
}, 1000)
})
}
当设置并发请求数量为 2 时 。
当设置并发请求数量为 3 时 。
p-limit 限制并发请求数量本质上是,在内部维护了一个请求队列; 。
当请求发起时,先将请求推入队列,判断当前执行的请求数量是否小于配置的请求并发数量,如果是则执行当前请求,否则等待正在发起的请求中谁请求完了,再从队列首部取出一个执行; 。
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); // 将结果传递给 generator
result.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 表示执行的请求, 。
resolve 由 generator 定义并往下传,一直跟踪到请求执行完毕后,调用 resolve(result); 代表 generator 函数 fulfilled 。
···args 表示其余的参数,最终会作为 fn 的参数.
4、执行 run 函数时 。
const run = (fn, resolve, ...args) => { // 请求开始执行
activeCount++; // 请求开始执行,当前请求数量 +1
const 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 兑现后,generator 的 promise 也就兑现了( 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 版本中间加了一些改进 。
在 3.0.0 中,作者将请求入队放在前面,将 if 判断语句和请求执行置于微任务中运行;正如源码注释中解释的:因为当 run 函数执行时,activeCount 是异步更新的,那么这里的 if 判断语句也应该异步执行才能实时获取到 activeCount 的值.
这样一开始批量执行 limit(fn) 时,将会先把这些请求全部放入队列中,然后再根据条件判断是否执行请求; 。
将 return Promise.reject 改为了直接 throw 一个错误 。
pTry
的依赖;改善性能;移除了 pTry 依赖,改为了 async 包裹,上面有提到,pTry 是一个 promise 包装函数,返回结果是一个 promise;两者本质都是一样; 。
增加了 yocto-queue 依赖,yocto-queue是一个队列数据结构,用队列代替数组,性能更好;队列的入队和出队操作时间复杂度是 O(1),而数组的 shift() 是 O(n),
引入了 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 在这里 。
移除了 AsyncResource.bind(),改为使用一个立即执行的 promise,并将 promise 的 resolve 方法插入队列,一旦 resolve 完成兑现,调用相应请求;相关 issue 在这里 。
改变并发数后立马再检测是否可以执行请求; 。
在上面第4点的,第5点中的优化没太看明白,因为执行请求用的到三个参数(fn,resolve,args)都是通过函数传参过来的,看起来 this 没关系,为啥要进行多层 bind 绑定呢?各位知道的可以不吝赐教下么.
最后此篇关于控制请求并发数量:p-limit源码解读的文章就讲到这里了,如果你想了解更多关于控制请求并发数量:p-limit源码解读的内容请搜索CFSDN的文章或继续浏览相关文章,希望大家以后支持我的博客! 。
我正在尝试在多线程环境中实现某种累积逻辑;我想知道没有 lock 和 synchronized 关键字是否有更好/更快的方法来做到这一点?以下是我当前的代码: public class Concurr
我需要帮助构建一个实现信号量的监视器,简单的 C 示例就可以。 这是为了证明可以在任何可以使用信号量的地方使用监视器。 最佳答案 如果您说允许使用互斥锁/condvars,请检查: #include
我已经构建了一些返回部分产品目录的 ajax,并且我正在尝试将 xml 输出到文档中,到目前为止,这是我所拥有的: $("#catalog").append("Item NamePriceDe
很抱歉,如果我的问题之前已经被问过,或者它太明显了,但我真的需要澄清这一点。感谢您的帮助。 在多用户界面中,如果来自不同用户的相同事务同时到达服务器,会发生什么? 我有下一张表: create tab
这可能是一个愚蠢的问题,但是这个程序的输出(它的方式)可以为零吗? public class Test2{ int a = 0; AtomicInteger b = new Atomi
假设我本地主机上的一个网站处理每个请求大约需要 3 秒。这很好,正如预期的那样(因为它在幕后进行了一些奇特的网络)。 但是,如果我在选项卡(在 firefox 中)中打开相同的 url,然后同时重新加
我对 MongoDB 的读锁定有点困惑。单个集合可以支持多少个并发读取操作? 最佳答案 如 tk 给出的链接中所写:http://www.mongodb.org/pages/viewpage.acti
如果有四个并发的 CUDA 应用程序在一个 GPU 中竞争资源会发生什么这样他们就可以将工作卸载到图形卡上了? Cuda Programming Guide 3.1 提到那里 某些方法是异步的: 内核
👊上次的百度面试遇到了关于spark的并发数的问题,今天我们就来将这些问题都一并解决一下,图画的的有点丑,还行大家见谅,百度实习的问题我放在了下面的链接👇: 链接: 2022百度大数据开发工程师实
我对 Groovy 线程有疑问。 我的任务是以某种方式翻译给定目录中的每个文件 并将生成的输出放在其他目录中的文件中。 我编写了以下代码,该代码有效: static def translateDir(
Java中的同步和锁定有什么区别? 最佳答案 synchronized是语言关键字;锁是对象。 当一个方法或代码块被标记为同步时,您是说该方法或代码块必须先获得某个锁对象(可以在同步的语法中指定)才能
我需要创建一个能够同时处理来自客户端的多个请求的并发 RPC 服务器。 使用 rpcgen linux编译器(基于sun RPC),不支持-A为并发服务器创建 stub 的选项。 (-A 选项在 so
System.out.println("Enter the number of what you would like to do"); System.out.println("1 = Manuall
我正在将我的应用程序移植到 iOS 8.0 并注意到 UIAlertView 已被弃用。 所以我改变了使用 UIAlertController 的方法。这在大多数情况下都有效。 除了,当我的应用程序打
我正在逐行同时读取两个文本文件。 我特别想做的是当lineCount在每个线程上都是相同的我想看看扫描仪当前正在读取的字符串。 我环顾四周寻找可以实现的某些模式,例如 Compare and Swap
我正在阅读 Java Concurrency in Practice .在章节中断政策部分 取消和关闭 它提到 A task should not assume anything about the
我正在尝试学习线程,互斥等的基础知识。遵循here的文档和示例。在下面的代码中,我得到预期的输出。问题: 想确认我是否有任何陷阱?我们如何改善下面的代码? 我的线程在哪一行尝试获取互斥锁或正在等待互斥
并发是指两个任务在不同的线程上并行运行。但是,异步方法并行运行,但在同一个线程上。这是如何实现的?另外,并行性怎么样? 这三个概念有什么区别? 最佳答案 并发和并行实际上与您正确推测的原理相同,两者都
以此ConcurrentDouble类定义为例: public class ConcurrentDouble { public double num = 0; public void subt
在得知并发确实增加了许多人的吞吐量后,我一直计划在项目中使用并发。现在我在多线程或并发方面还没有做太多工作,因此决定在实际项目中使用它之前学习并进行简单的概念验证。 以下是我尝试过的两个示例: 1.
我是一名优秀的程序员,十分优秀!