最近,阿宝哥在梳理 CLI(Command Line Interface)的相关内容,就对优秀的 Lerna 产生了兴趣,于是开始 “啃” 起了它的源码。在阅读开源项目时,阿宝哥习惯先阅读项目的 「README.md」 文档和 「package.json」 文件,而在 「package.json」 文件的 「dependencies」 字段中,阿宝哥见到了多个 「p-」* 的依赖包:
{
"name": "lerna-monorepo",
"version": "4.0.0-monorepo",
"dependencies": {
"p-map": "^4.0.0",
"p-map-series": "^2.1.0",
"p-pipe": "^3.1.0",
"p-queue": "^6.6.2",
"p-reduce": "^2.1.0",
"p-waterfall": "^2.1.1"
}
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
提示:如果你想知道阿宝哥如何阅读开源项目的话,可以阅读 使用这些思路与技巧,我读懂了多个优秀的开源项目 这篇文章。
之后,阿宝哥顺藤摸瓜找到了 promise-fun (3.5K) 这个项目。该项目的作者 「sindresorhus」 是一个全职做开源的大牛,Github 上拥有 「43.9K」 的关注者。同时,他还维护了多个优秀的开源项目,比如 awesome (167K)、awesome-nodejs (42K)、got (9.8K)、ora (7.1K) 和 screenfull.js (6.1K) 等项目。
(图片来源 —— https://github.com/sindresorhus)
promise-fun 项目收录了 「sindresorhus」 写过的 「48」 个与 Promise 相关的模块,比如前面见到的 「p-map」、「p-map-series」、「p-pipe」 和 「p-waterfall」 等模块。本文阿宝哥将筛选一些比较常用的模块,来详细分析每个模块的用法和具体的代码实现。
这些模块提供了很多有用的方法,利用这些方法,可以帮我们解决工作中一些很常见的问题,比如实现并发控制、异步任务处理等,特别是处理多种控制流,比如 「series」、「waterfall」、「all」、「race」 和 「forever」 等。
初始化示例项目
创建一个新的 「learn-promise-fun」 项目,然后在该项目的根目录下,执行 npm init -y 命令进行项目初始化操作。当该命令成功运行后,在项目根目录下将会生成 「package.json」 文件。由于 promise-fun 项目中的很多模块使用了 ES Module,为了保证后续的示例代码能够正常运行,我们需要在 「package.json」 文件中,增加 「type」 字段并设置它的值为 「"module"」。
由于阿宝哥本地 Node.js 的版本是 「v12.16.2」,要运行 ES Module 模块,还要添加 「--experimental-modules」 命令行选项。而如果你不想看到警告消息,还可以添加 「--no-warnings」 命令行选项。此外,为了避免每次运行示例代码时,都需要输入以上命令行选项,我们可以在 「package.json」 的 「scripts」 字段中定义相应的 「npm script」 命令,具体如下所示:
{
"name": "learn-promise-fun",
"type": "module",
"scripts": {
"pall": "node --experimental-modules ./p-all/p-all.test.js",
"pfilter": "node --experimental-modules ./p-filter/p-filter.test.js",
"pforever": "node --experimental-modules ./p-forever/p-forever.test.js",
"preduce": "node --experimental-modules ./p-reduce/p-reduce.test.js",
...
},
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
在完成项目初始化之后,我们先来回顾一下大家平时用得比较多的 reduce、map 和 filter 数组方法的特点:
提示:上图通过👉 「https://carbon.now.sh/」 在线网页制作生成。
相信大家对图中的 「Array.prototype.reduce」 方法不会陌生,该方法用于对数组中的每个元素执行一个 「reducer」 函数,并将结果汇总为单个返回值。对应的使用示例,如下所示:
const array1 = [1, 2, 3, 4];
const reducer = (accumulator, currentValue) => accumulator + currentValue;
// 1 + 2 + 3 + 4
console.log(array1.reduce(reducer)); // 10
- 1.
- 2.
- 3.
- 4.
- 5.
其中 「reducer」 函数接收 4 个参数:
- acc(Accumulator):累计器
- cur(Current Value):当前值
- idx(Current Index):当前索引
- src(Source Array):源数组
而接下来,我们要介绍的 p-reduce 模块,就提供了跟 「Array.prototype.reduce」 方法类似的功能。
p-reduce
Reduce a list of values using promises into a promise for a value
https://github.com/sindresorhus/p-reduce
使用说明p-reduce 适用于需要根据异步资源计算累加值的场景。该模块默认导出了一个 「pReduce」 函数,该函数的签名如下:
「pReduce(input, reducer, initialValue): Promise」
- input: Iterable<Promise|any>
- reducer(previousValue, currentValue, index): Function
- initialValue: unknown
使用示例
// p-reduce/p-reduce.test.js
import delay from "delay";
import pReduce from "p-reduce";
const inputs = [Promise.resolve(1), delay(50, { value: 6 }), 8];
async function main() {
const result = await pReduce(inputs, async (a, b) => a + b, 0);
console.dir(result); // 输出结果:15
}
main();
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
在以上示例中,我们导入了 「delay」 模块默认导出的 delay 方法,该方法可用于按照给定的时间,延迟一个 Promise 对象。即在给定的时间之后,Promise 状态才会变成 「resolved」。默认情况下,「delay」 模块内部是通过 setTimeout API 来实现延迟功能的。示例中 delay(50, { value: 6 }) 表示延迟 50ms 后,Promise 对象的返回值为 「6」。而在 main 函数内部,我们使用了 pReduce 函数来计算 inputs 数组元素的累加值。当以上代码成功运行之后,命令行会输出 「15」。
下面我们来分析一下 pReduce 函数内部是如何实现的。
源码分析
// https://github.com/sindresorhus/p-reduce/blob/main/index.js
export default async function pReduce(iterable, reducer, initialValue) {
return new Promise((resolve, reject) => {
const iterator = iterable[Symbol.iterator](); // 获取迭代器
let index = 0; // 索引值
const next = async (total) => {
const element = iterator.next(); // 获取下一项
if (element.done) { // 判断迭代器是否迭代完成
resolve(total);
return;
}
try {
const [resolvedTotal, resolvedValue] = await Promise.all([
total,
element.value,
]);
// 迭代下一项
// reducer(previousValue, currentValue, index): Function
next(reducer(resolvedTotal, resolvedValue, index++));
} catch (error) {
reject(error);
}
};
// 使用初始值,开始迭代
next(initialValue);
});
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
- 20.
- 21.
- 22.
- 23.
- 24.
- 25.
- 26.
- 27.
- 28.
- 29.
- 30.
- 31.
在以上代码中,核心的流程就是通过获取 iterable 对象内部的迭代器,来不断地进行迭代。此外,在 pReduce 函数中,使用了 「Promise.all」 方法,该方法会返回一个 promise 对象,当输入的所有 promise 对象的状态都变成 resolved 时,返回的 promise 对象就会以数组的形式,返回每个 promise 对象 resolve 后的结果。当输入的任何一个 promise 对象状态变成 rejected 时,则返回的 promise 对象会 reject 对应的错误信息。
不过,需要注意的是,「Promise.all」 方法存在兼容性问题,具体的兼容性如下图所示:
(图片来源 —— https://caniuse.com/?search=Promise.all)
可能有一些小伙伴对 「Promise.all」 还不熟悉,它又是一道比较高频的手写题。所以,接下来我们来手写一个简易版的 「Promise.all」:
Promise.all = function (iterators) {
return new Promise((resolve, reject) => {
if (!iterators || iterators.length === 0) {
resolve([]);
} else {
let count = 0; // 计数器,用于判断所有任务是否执行完成
let result = []; // 结果数组
for (let i = 0; i < iterators.length; i++) {
// 考虑到iterators[i]可能是普通对象,则统一包装为Promise对象
Promise.resolve(iterators[i]).then(
(data) => {
result[i] = data; // 按顺序保存对应的结果
// 当所有任务都执行完成后,再统一返回结果
if (++count === iterators.length) {
resolve(result);
}
},
(err) => {
reject(err); // 任何一个Promise对象执行失败,则调用reject()方法
return;
}
);
}
}
});
};
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
- 20.
- 21.
- 22.
- 23.
- 24.
- 25.
- 26.
p-map
Map over promises concurrently
https://github.com/sindresorhus/p-map
使用说明
p-map 适用于使用不同的输入多次运行 「promise-returning」 或 「async」 函数的场景。与前面介绍的 Promise.all 方法的区别是,你可以控制并发,也可以决定是否在出现错误时停止迭代。该模块默认导出了一个 「pMap」 函数,该函数的签名如下:
「pMap(input, mapper, options): Promise」
- input: Iterable<Promise | unknown>
-
mapper(element, index): Function -
options: object
stopOnError: boolean —— 出现异常时,是否终止,默认值为 true。
了解完 「pMap」 函数的签名之后,我们来看一下该函数如何使用。
使用示例
// p-map/p-map.test.js
import delay from "delay";
import pMap from "p-map";
const inputs = [200, 100, 50];
const mapper = (value) => delay(value, { value });
async function main() {
console.time("start");
const result = await pMap(inputs, mapper, { concurrency: 1 });
console.dir(result); // 输出结果:[ 200, 100, 50 ]
console.timeEnd("start");
}
main();
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
在以上示例中,我们也使用了 「delay」 模块导出的 delay 函数,用于实现延迟功能。当成功执行以上代码后,命令行会输出以下信息:
[ 200, 100, 50 ]
start: 368.708ms
- 1.
- 2.
而当把 concurrency 属性的值更改为 2 之后,再次执行以上代码。那么命令行将会输出以下信息:
[ 200, 100, 50 ]
start: 210.322ms
- 1.
- 2.
观察以上的输出结果,我们可以看出并发数为 1 时,程序的运行时间大于 350ms。而如果并发数为 2 时,多个任务是并行执行的,所以程序的运行时间只需 210 多毫秒。那么 pMap 函数,内部是如何实现并发控制的呢?下面来分析一下 pMap 函数的源码。
源码分析
// https://github.com/sindresorhus/p-map/blob/main/index.js
import AggregateError from "aggregate-error";
export default async function pMap(
iterable,
mapper,
{ concurrency = Number.POSITIVE_INFINITY, stopOnError = true } = {}
) {
return new Promise((resolve, reject) => {
// 省略参数校验代码
const result = []; // 存储返回结果
const errors = []; // 存储异常对象
const skippedIndexes = []; // 保存跳过项索引值的数组
const iterator = iterable[Symbol.iterator](); // 获取迭代器
let isRejected = false; // 标识是否出现异常
let isIterableDone = false; // 标识是否已迭代完成
let resolvingCount = 0; // 正在处理的任务个数
let currentIndex = 0; // 当前索引
const next = () => {
if (isRejected) { // 若出现异常,则直接返回
return;
}
const nextItem = iterator.next(); // 获取下一项
const index = currentIndex; // 记录当前的索引值
currentIndex++;
if (nextItem.done) { // 判断迭代器是否迭代完成
isIterableDone = true;
// 判断是否所有的任务都已经完成了
if (resolvingCount === 0) {
if (!stopOnError && errors.length > 0) { // 异常处理
reject(new AggregateError(errors));
} else {
for (const skippedIndex of skippedIndexes) {
// 删除跳过的值,不然会存在空的占位
result.splice(skippedIndex, 1);
}
resolve(result); // 返回最终的处理结果
}
}
return;
}
resolvingCount++; // 正在处理的任务数加1
(async () => {
try {
const element = await nextItem.value;
if (isRejected) {
return;
}
// 调用mapper函数,进行值进行处理
const value = await mapper(element, index);
// 处理跳过的情形,可以在mapper函数中返回pMapSkip,来跳过当前项
// 比如在异常捕获的catch语句中,返回pMapSkip值
if (value === pMapSkip) { // pMapSkip = Symbol("skip")
skippedIndexes.push(index);
} else {
result[index] = value; // 把返回值按照索引进行保存
}
resolvingCount--;
next(); // 迭代下一项
} catch (error) {
if (stopOnError) { // 出现异常时,是否终止,默认值为true
isRejected = true;
reject(error);
} else {
errors.push(error);
resolvingCount--;
next();
}
}
})();
};
// 根据配置的concurrency值,并发执行任务
for (let index = 0; index < concurrency; index++) {
next();
if (isIterableDone) {
break;
}
}
});
}
export const pMapSkip = Symbol("skip");
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
- 20.
- 21.
- 22.
- 23.
- 24.
- 25.
- 26.
- 27.
- 28.
- 29.
- 30.
- 31.
- 32.
- 33.
- 34.
- 35.
- 36.
- 37.
- 38.
- 39.
- 40.
- 41.
- 42.
- 43.
- 44.
- 45.
- 46.
- 47.
- 48.
- 49.
- 50.
- 51.
- 52.
- 53.
- 54.
- 55.
- 56.
- 57.
- 58.
- 59.
- 60.
- 61.
- 62.
- 63.
- 64.
- 65.
- 66.
- 67.
- 68.
- 69.
- 70.
- 71.
- 72.
- 73.
- 74.
- 75.
- 76.
- 77.
- 78.
- 79.
- 80.
- 81.
- 82.
- 83.
- 84.
- 85.
- 86.
- 87.
- 88.
- 89.
- 90.
- 91.
- 92.
pMap 函数内部的处理逻辑还是蛮清晰的,把核心的处理逻辑都封装在 next 函数中。在调用 pMap 函数之后,内部会根据配置的concurrency 值,并发调用 next 函数。而在 next 函数内部,会通过 「async/await」 来控制任务的执行过程。
在 pMap 函数中,作者巧妙设计了 「pMapSkip」。当我们在 mapper 函数中返回了 「pMapSkip」 之后,将会从返回的结果数组中移除对应索引项的值。了解完 「pMapSkip」 的作用之后,我们来举个简单的例子:
import pMap, { pMapSkip } from "p-map";
const inputs = [200, pMapSkip, 50];
const mapper = async (value) => value;
async function main() {
console.time("start");
const result = await pMap(inputs, mapper, { concurrency: 2 });
console.dir(result); // [ 200, 50 ]
console.timeEnd("start");
}
main();
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
在以上代码中,我们的 inputs 数组中包含了 pMapSkip 值,当使用 pMap 函数对 inputs 数组进行处理后,pMapSkip 值将会被过滤掉,所以最终 result 的结果为 「[200 , 50]」。
p-filter
Filter promises concurrently
https://github.com/sindresorhus/p-filter
使用说明p-filter 适用于使用不同的输入多次运行 「promise-returning」 或 「async」 函数,并对返回的结果进行过滤的场景。该模块默认导出了一个 「pFilter」 函数,该函数的签名如下:
「pFilter(input, filterer, options): Promise」
- input: Iterable<Promise | any>
-
filterer(element, index): Function -
options: object
了解完 「pFilter」 函数的签名之后,我们来看一下该函数如何使用。
使用示例
// p-filter/p-filter.test.js
import pFilter from "p-filter";
const inputs = [Promise.resolve(1), 2, 3];
const filterer = (x) => x % 2;
async function main() {
const result = await pFilter(inputs, filterer, { concurrency: 1 });
console.dir(result); // 输出结果:[ 1, 3 ]
}
main();
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
在以上示例中,我们使用 pFilter 函数对包含 Promise 对象的 inputs 数组,应用了 (x) => x % 2 过滤器。当以上代码成功运行后,命令行会输出 「[1, 3]」。
源码分析
// https://github.com/sindresorhus/p-filter/blob/main/index.js
const pMap = require('p-map');
const pFilter = async (iterable, filterer, options) => {
const values = await pMap(
iterable,
(element, index) => Promise.all([filterer(element, index), element]),
options
);
return values.filter(value => Boolean(value[0])).map(value => value[1]);
};
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
由以上代码可知,在 pFilter 函数内部,使用了我们前面已经介绍过的 pMap 和 Promise.all 函数。要理解以上代码,我们还需要来回顾一下 pMap 函数的关键代码:
// https://github.com/sindresorhus/p-map/blob/main/index.js
export default async function pMap(
iterable, mapper,
{ concurrency = Number.POSITIVE_INFINITY, stopOnError = true } = {}
) {
const iterator = iterable[Symbol.iterator](); // 获取迭代器
let currentIndex = 0; // 当前索引
const next = () => {
const nextItem = iterator.next(); // 获取下一项
const index = currentIndex;
currentIndex++;
(async () => {
try {
// element => await Promise.resolve(1);
const element = await nextItem.value;
// mapper => (element, index) => Promise.all([filterer(element, index), element])
const value = await mapper(element, index);
if (value === pMapSkip) {
skippedIndexes.push(index);
} else {
result[index] = value; // 把返回值按照索引进行保存
}
resolvingCount--;
next(); // 迭代下一项
} catch (error) {
// 省略异常处理代码
}
})();
}
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
- 20.
- 21.
- 22.
- 23.
- 24.
- 25.
- 26.
- 27.
- 28.
- 29.
- 30.
- 31.
因为 pFilter 函数中所用的 mapper 函数是 (element, index) => Promise.all([filterer(element, index), element]),所以 await mapper(element, index) 表达式的返回值是一个数组。数组的第 1 项是 filterer 过滤器的处理结果,而数组的第 2 项是当前元素的值。因此在调用 pMap 函数后,它的返回值是一个二维数组。所以在获取 pMap 函数的返回值之后, 会使用以下语句对返回值进行处理:
values.filter(value => Boolean(value[0])).map(value => value[1])
- 1.
其实,对于前面的 pFilter 示例来说,除了 inputs 可以含有 Promise 对象,我们的 filterer 过滤器也可以返回 Promise 对象:
import pFilter from "p-filter";
const inputs = [Promise.resolve(1), 2, 3];
const filterer = (x) => Promise.resolve(x % 2);
async function main() {
const result = await pFilter(inputs, filterer);
console.dir(result); // [ 1, 3 ]
}
main();
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
以上代码成功执行后,命令行的输出结果也是 「[1, 3]」。好的,现在我们已经介绍了 p-reduce、p-map 和 p-filter 3 个模块。下面我们来继续介绍另一个模块 —— p-waterfall。
p-waterfall
Run promise-returning & async functions in series, each passing its result to the next
https://github.com/sindresorhus/p-waterfall
使用说明
p-waterfall 适用于串行执行 「promise-returning」 或 「async」 函数,并把前一个函数的返回结果自动传给下一个函数的场景。该模块默认导出了一个 「pWaterfall」 函数,该函数的签名如下:
「pWaterfall(tasks, initialValue): Promise」
- tasks: Iterable<Function>
initialValue: unknown:将作为第一个任务的 previousValue
使用示例
// p-waterfall/p-waterfall.test.js
import pWaterfall from "p-waterfall";
const tasks = [
async (val) => val + 1,
(val) => val + 2,
async (val) => val + 3,
];
async function main() {
const result = await pWaterfall(tasks, 0);
console.dir(result); // 输出结果:6
}
main();
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
在以上示例中,我们创建了 3 个任务,然后使用 pWaterfall 函数来执行这 3 个任务。当以上代码成功执行后,命令行会输出 「6」。对应的执行流程如下图所示:
源码分析
// https://github.com/sindresorhus/p-waterfall/blob/main/index.js
import pReduce from 'p-reduce';
export default async function pWaterfall(iterable, initialValue) {
return pReduce(iterable, (previousValue, function_) => function_(previousValue), initialValue);
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
在 pWaterfall 函数内部,会利用前面已经介绍的 pReduce 函数来实现 「waterfall」 流程控制。同样,要搞清楚内部的控制流程,我们需要来回顾一下 pReduce 函数的具体实现:
export default async function pReduce(iterable, reducer, initialValue) {
return new Promise((resolve, reject) => {
const iterator = iterable[Symbol.iterator](); // 获取迭代器
let index = 0; // 索引值
const next = async (total) => {
const element = iterator.next(); // 获取下一项
if (element.done) {
// 判断迭代器是否迭代完成
resolve(total);
return;
}
try {
// 首次调用next函数的状态:
// resolvedTotal => 0
// element.value => async (val) => val + 1
const [resolvedTotal, resolvedValue] = await Promise.all([
total,
element.value,
]);
// reducer => (previousValue, function_) => function_(previousValue)
next(reducer(resolvedTotal, resolvedValue, index++));
} catch (error) {
reject(error);
}
};
// 使用初始值,开始迭代
next(initialValue);
});
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
- 20.
- 21.
- 22.
- 23.
- 24.
- 25.
- 26.
- 27.
- 28.
- 29.
- 30.
- 31.
- 32.
- 33.
现在我们已经知道 「pWaterfall」 函数会把前一个任务的输出结果,作为输入传给下一个任务。但有些时候,在串行执行每个任务时,我们并不关心每个任务的返回值。针对这种场合,我们可以考虑使用 p-series 模块提供的 pSeries 函数。
p-series
Run promise-returning & async functions in series
https://github.com/sindresorhus/p-series
使用说明
p-series 适用于串行执行 「promise-returning」 或 「async」 函数的场景。
使用示例
// p-series/p-series.test.js
import pSeries from "p-series";
const tasks = [async () => 1 + 1, () => 2 + 2, async () => 3 + 3];
async function main() {
const result = await pSeries(tasks);
console.dir(result); // 输出结果:[2, 4, 6]
}
main();
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
在以上示例中,我们创建了 3 个任务,然后使用 pSeries 函数来执行这 3 个任务。当以上代码成功执行后,命令行会输出 「[ 2, 4, 6 ]」。对应的执行流程如下图所示:
源码分析
// https://github.com/sindresorhus/p-series/blob/main/index.js
export default async function pSeries(tasks) {
for (const task of tasks) {
if (typeof task !== 'function') {
throw new TypeError(`Expected task to be a \`Function\`, received \`${typeof task}\``);
}
}
const results = [];
for (const task of tasks) {
results.push(await task()); // eslint-disable-line no-await-in-loop
}
return results;
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
由以上代码可知,在 pSeries 函数内部是利用 「for...of」 语句和 「async/await」 特性来实现串行任务流控制。因此在实际的项目中,你也可以无需使用该模块,就可以轻松的实现串行任务流控制。
p-all
Run promise-returning & async functions concurrently with optional limited concurrency
https://github.com/sindresorhus/p-all
使用说明
p-all 适用于并发执行 「promise-returning」 或 「async」 函数的场景。该模块提供的功能,与 「Promise.all」 API 类似,主要的区别是该模块允许你限制任务的并发数。在日常开发过程中,一个比较常见的场景就是控制 HTTP 请求的并发数,这时你也可以考虑使用 async-pool 这个库来解决并发控制的问题,如果你对该库的内部实现原理感兴趣的话,可以阅读 JavaScript 中如何实现并发控制? 这篇文章。
下面我们来继续介绍 p-all 模块,该模块默认导出了一个 「pAll」 函数,该函数的签名如下:
「pAll(tasks, options)」
- tasks: Iterable<Function>
options: object
stopOnError: boolean —— 出现异常时,是否终止,默认值为 true。
使用示例
// p-all/p-all.test.js
import delay from "delay";
import pAll from "p-all";
const inputs = [
() => delay(200, { value: 1 }),
async () => {
await delay(100);
return 2;
},
async () => 8,
];
async function main() {
console.time("start");
const result = await pAll(inputs, { concurrency: 1 });
console.dir(result); // 输出结果:[ 1, 2, 8 ]
console.timeEnd("start");
}
main();
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
- 20.
- 21.
在以上示例中,我们创建了 3 个异步任务,然后通过 pAll 函数来执行已创建的任务。当成功执行以上代码后,命令行会输出以下信息:
[ 1, 2, 8 ]
start: 312.561ms
- 1.
- 2.
而当把 concurrency 属性的值更改为 2 之后,再次执行以上代码。那么命令行将会输出以下信息:
[ 1, 2, 8 ]
start: 209.469ms
- 1.
- 2.
可以看出并发数为 1 时,程序的运行时间大于 300ms。而如果并发数为 2 时,前面两个任务是并行的,所以程序的运行时间只需 200 多毫秒。
源码分析
// https://github.com/sindresorhus/p-all/blob/main/index.js
import pMap from 'p-map';
export default async function pAll(iterable, options) {
return pMap(iterable, element => element(), options);
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
很明显在 pAll 函数内部,是通过 p-map 模块提供的 pMap 函数来实现并发控制的。如果你对 pMap 函数的内部实现方式,还不清楚的话,可以回过头再次阅读 p-map 模块的相关内容。接下来,我们来继续介绍另一个模块 —— p-race。
p-race
A better Promise.race()https://github.com/sindresorhus/p-race
使用说明
p-race 这个模块修复了 Promise.race API 一个 “愚蠢” 的行为。当使用空的可迭代对象,调用 Promise.race API 时,将会返回一个永远处于 「pending」 状态的 Promise 对象,这可能会产生一些非常难以调试的问题。而如果往 p-race 模块提供的 pRace 函数中传入一个空的可迭代对象时,该函数将会立即抛出 「RangeError: Expected the iterable to contain at least one item」 的异常信息。
「pRace(iterable)」 方法会返回一个 promise 对象,一旦迭代器中的某个 promise 对象 「resolved」 或 「rejected」,返回的 promise 对象就会 resolve 或 reject 相应的值。
使用示例
// p-race/p-race.test.js
import delay from "delay";
import pRace from "p-race";
const inputs = [delay(50, { value: 1 }), delay(100, { value: 2 })];
async function main() {
const result = await pRace(inputs);
console.dir(result); // 输出结果:1
}
main();
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
在以上示例中,我们导入了 「delay」 模块默认导出的 delay 方法,该方法可用于按照给定的时间,延迟一个 Promise 对象。利用 delay 函数,我们创建了 2 个 Promise 对象,然后使用 pRace 函数来处理这两个 Promise 对象。以上代码成功运行后,命令行始终会输出 「1」。那么为什么会这样呢?下面我们来分析一下 pRace 函数的源码。
源码分析
// https://github.com/sindresorhus/p-race/blob/main/index.js
import isEmptyIterable from 'is-empty-iterable';
export default async function pRace(iterable) {
if (isEmptyIterable(iterable)) {
throw new RangeError('Expected the iterable to contain at least one item');
}
return Promise.race(iterable);
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
观察以上源码可知,在 pRace 函数内部会先判断传入的 iterable 参数是否为空的可迭代对象。检测参数是否为空的可迭代对象,是通过 isEmptyIterable 函数来实现,该函数的具体代码如下所示:
// https://github.com/sindresorhus/is-empty-iterable/blob/main/index.js
function isEmptyIterable(iterable) {
for (const _ of iterable) {
return false;
}
return true;
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
当发现是空的可迭代对象时,pRace 函数会直接抛出 RangeError 异常。否则,会利用 Promise.race API 来实现具体的功能。需要注意的是,「Promise.race」 方法也存在兼容性问题,具体如下图所示:
(图片来源 —— https://caniuse.com/?search=Promise.race)
同样,可能有一些小伙伴对 「Promise.race」 还不熟悉,它也是一道挺高频的手写题。所以,接下来我们来手写一个简易版的 「Promise.race」:
Promise.race = function (iterators) {
return new Promise((resolve, reject) => {
for (const iter of iterators) {
Promise.resolve(iter)
.then((res) => {
resolve(res);
})
.catch((e) => {
reject(e);
});
}
});
};
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
p-forever
Run promise-returning & async functions repeatedly until you end it
https://github.com/sindresorhus/p-forever
使用说明
p-forever 适用于需要重复不断执行 「promise-returning」 或 「async」 函数,直到用户终止的场景。该模块默认导出了一个 「pForever」 函数,该函数的签名如下:
「pForever(fn, initialValue)」
- fn: Function:重复执行的函数;
- initialValue:传递给 fn 函数的初始值。
了解完 「pForever」 函数的签名之后,我们来看一下该函数如何使用。
使用示例
// p-forever/p-forever.test.js
import delay from "delay";
import pForever from "p-forever";
async function main() {
let index = 0;
await pForever(async () => (++index === 10 ? pForever.end : delay(50)));
console.log("当前index的值: ", index); // 输出结果:当前index的值: 10
}
main();
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
在以上示例中,传入 pForever 函数的 fn 函数会一直重复执行,直到该 fn 函数返回 pForever.end 的值,才会终止执行。因此以上代码成功执行后,命令行的输出结果是:「当前index的值: 10」。
源码分析
// https://github.com/sindresorhus/p-forever/blob/main/index.js
const endSymbol = Symbol('pForever.end');
const pForever = async (function_, previousValue) => {
const newValue = await function_(await previousValue);
if (newValue === endSymbol) {
return;
}
return pForever(function_, newValue);
};
pForever.end = endSymbol;
export default pForever;
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
由以上源码可知,pForever 函数的内部实现并不复杂。当判断 newValue 的值为 endSymbol 时,就直接返回了。否则,就会继续调用 pForever 函数。除了一直重复执行任务之外,有时候我们会希望显式指定任务的执行次数,针对这种场景,我们就可以使用 p-times 模块。
p-times
Run promise-returning & async functions a specific number of times concurrently
https://github.com/sindresorhus/p-times
使用说明
p-times 适用于显式指定 「promise-returning」 或 「async」 函数执行次数的场景。该模块默认导出了一个 「pTimes」 函数,该函数的签名如下:
「pTimes(count, mapper, options): Promise」
- count: number:调用次数;
- mapper(index): Function:mapper 函数,调用该函数后会返回 Promise 对象或某个具体的值;
- options: object
concurrency: number —— 并发数,默认值 Infinity,最小值为 1;
stopOnError: boolean —— 出现异常时,是否终止,默认值为 true。
了解完 「pTimes」 函数的签名之后,我们来看一下该函数如何使用。
使用示例
// p-times/p-times.test.js
import delay from "delay";
import pTimes from "p-times";
async function main() {
console.time("start");
const result = await pTimes(5, async (i) => delay(50, { value: i * 10 }), {
concurrency: 3,
});
console.dir(result);
console.timeEnd("start");
}
main();
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
在以上示例中,我们通过 pTimes 函数配置 「mapper」 函数的执行次数为 「5」 次,同时设置任务的并发数为 「3」。当以上代码成功运行后,命令行会输出以下结果:
[ 0, 10, 20, 30, 40 ]
start: 116.090ms
- 1.
- 2.
对于以上示例,你可以通过改变 concurrency 的值,来对比输出的程序运行时间。那么 pTimes 函数内部是如何实现并发控制的呢?其实该函数内部也是利用 pMap 函数来实现并发控制。
源码分析
// https://github.com/sindresorhus/p-times/blob/main/index.js
import pMap from "p-map";
export default function pTimes(count, mapper, options) {
return pMap(
Array.from({ length: count }).fill(),
(_, index) => mapper(index),
options
);
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
在 pTimes 函数中,会通过 Array.from 方法创建指定长度的数组,然后通过 fill 方法进行填充。最后再把该数组、mapper 函数和 options 配置对象,作为输入参数调用 pMap 函数。写到这里,阿宝哥觉得 pMap 函数提供的功能还是蛮强大的,很多模块的内部都使用了 pMap 函数。
p-pipe
Compose promise-returning & async functions into a reusable pipeline
https://github.com/sindresorhus/p-pipe
使用说明
p-pipe 适用于把 「promise-returning」 或 「async」 函数组合成可复用的管道。该模块默认导出了一个 「pPipe」 函数,该函数的签名如下:
「pPipe(input...)」
- input: Function:期望调用后会返回 Promise 或任何值的函数。
了解完 「pPipe」 函数的签名之后,我们来看一下该函数如何使用。
使用示例
// p-pipe/p-pipe.test.js
import pPipe from "p-pipe";
const addUnicorn = async (string) => `${string} Unicorn`;
const addRainbow = async (string) => `${string} Rainbow`;
const pipeline = pPipe(addUnicorn, addRainbow);
(async () => {
console.log(await pipeline("❤️")); // 输出结果:❤️ Unicorn Rainbow
})();
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
在以上示例中,我们通过 pPipe 函数把 addUnicorn 和 addRainbow 这两个函数组合成一个新的可复用的管道。被组合函数的执行顺序是从左到右,所以以上代码成功运行后,命令行会输出 「❤️ Unicorn Rainbow」。
源码分析
// https://github.com/sindresorhus/p-pipe/blob/main/index.js
export default function pPipe(...functions) {
if (functions.length === 0) {
throw new Error('Expected at least one argument');
}
return async input => {
let currentValue = input;
for (const function_ of functions) {
currentValue = await function_(currentValue); // eslint-disable-line no-await-in-loop
}
return currentValue;
};
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
由以上代码可知,在 pPipe 函数内部是利用 「for...of」 语句和 「async/await」 特性来实现管道的功能。分析完 promise-fun 项目中的 10 个模块之后,再次感受到 「async/await」 特性给前端异步编程带来的巨大便利。其实对于异步的场景来说,除了可以使用 promise-fun 项目中收录的模块之外,还可以使用 async 或 neo-async 这两个异步处理模块提供的工具函数。在 Webpack 项目中,就用到了 neo-async 这个模块,该模块的作者是希望用来替换 async 模块,以提供更好的性能。建议需要经常处理异步场景的小伙伴,抽空浏览一下 neo-async 这个模块的 「官方文档」。
总结
promise-fun 项目共收录了 「50」 个与 Promise 有关的模块,该项目的作者 「sindresorhus」 个人就开发了 「48」 个模块,不愧是全职做开源的大牛。由于篇幅有限,阿宝哥只介绍了其中 「10」 个比较常用的模块。其实该项目还包含一些挺不错的模块,比如 「p-queue」、「p-any」、「p-some」、「p-debounce」、「p-throttle」 和 「p-timeout」 等。感兴趣的小伙伴,可以自行了解一下其他的模块。
参考资源
- MDN — Array.prototype.reduce()
- MDN — Array.from
- Promise.race with empty lists
- neo-async 官方文档