手写p-map（控制并发数以及迭代处理promise的库）-6ren

手写p-map（控制并发数以及迭代处理promise的库）

转载作者：撒哈拉更新时间：2024-10-08 22:18:58

介绍

p-map 是一个迭代处理 promise 并且能控制 promise 执行并发数的库。作者是 sindresorhus，他还创建了许多关于 promise 的库 promise-fun，感兴趣的同学可以去看看.

之前提到的 p-limit 也是一个控制请求并发数的库，控制并发数方面，两者作用相同，不过 p-map 增加了对请求（promise）的迭代处理.

之前 p-limit 的用法如下，limit 接受一个函数；。

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

而 p-map 则通过用户传进来的 mapper 处理函数处理的一个集合（准确的说是一个可迭代对象）；。

import pMap from 'p-map';
import got from 'got';

const sites = [
	getWebsiteFromUsername('sindresorhus'), //=> Promise
	'https://avajs.dev',
	'https://github.com'
];

const mapper = async site => {
	const {requestUrl} = await got.head(site);
	return requestUrl;
};

// 接收三个参数，一个是可迭代对象，一个是对可迭代对象进行处理的函数，一个是配置选项
const result = await pMap(sites, mapper, {concurrency: 2});

console.log(result);
//=> ['https://sindresorhus.com/', 'https://avajs.dev/', 'https://github.com/']

默认的可迭代对象有String、Array、TypedArray、Map、Set 、Intl.Segments，而要成为可迭代对象，该对象必须实现 [Symbol.iterator]() 方法；。

遍历可迭代对象时，实际上是根据迭代器协议进行遍历；。

比如，迭代一个数组是这样的。

var iterable = [1,2,3,4]
var iterator = iterable[Symbol.iterator]() // iterator 是迭代器
iterator.next() // {value: 1, done: false}
iterator.next() // {value: 2, done: false}
iterator.next() // {value: 3, done: false}
iterator.next() // {value: 4, done: false}
iterator.next() // {value: undefined, done: true}

当数组迭代完成后，会返回 {value: undefined, done: true} 。

p-pap 控制并发请求的原理是，对传进来的集合进行迭代，当集合第一个元素（元素可能是异步函数）执行完后，会交给 mapper 函数处理，mapper 处理完后，才开始迭代下一个元素，这样就保持了按照顺序一个个迭代，此时并发数是1；。

要做到并发是n，并且还能执行上面的迭代，作者很巧妙的用了 for 循环。

for(let i=0;i<n;i++)
    next()
}

手写 `p-map`

下面按照作者思路实现一个 p-map，现在有一个需要处理的可迭代对象 arr 。

var fetchSomething = (str,ms) =>{
    return new Promise((resolve,reject) =>{
        setTimeout(() =>{
            resolve(parseFloat(str))
        },ms)
    })
}

var arr= [
    fetchSomething('1a' ,1000), // promise
    2,3, 
    fetchSomething( '4a' , 5000), // promise
    5,6,7,8,9,10
]

集合第一个元素和第四个元素是一个 promise 函数。

p-map 接收三个参数，分别是要迭代的对象，mapper 处理函数，自定义配置；返回值是 promise，如下面所示。

var pMap = (iterable, mapper, options) => new Promise((resolve,reject) => {});

拿到可迭代对象后，对它进行递归迭代，直至迭代完毕；这里定义一个内部递归迭代函数 next 。

var pMap = (iterable, mapper, options) => new Promise((resolve,reject) => {
    var iterator = iterable[Symbol.iterator]()

    var next= ()=>{
        var item=iterator.next()
        if(item.done){
            return
        }
        next()
    }
    
});

迭代对象中每个元素都是按顺序迭代的；如果元素是异步函数时，需要先等异步函数兑现，并且兑现后的值传给 mapper 函数，等到 mapper 函数兑现或者拒绝后才继续迭代下一个元素。

var iterator = iterable[Symbol.iterator]()
var next = () => {
    var item = iterator.next()

    if (item.done) {
        return
    }

    Promise.resolve(item.value)
        .then(res => mapper(res))
        .then(res2 => {
            next()
        })
}

并且每次迭代彻底完成后保存兑现的结果。

var iterator = iterable[Symbol.iterator]()
var index = 0 // 序号，根据可迭代对象的顺序保存结果
var ret = []// 保存结果
var next = () => {
    var item = iterator.next()

    if (item.done) {
        return
    }

    var currentIndex = index //保存当前元素序号，用于存入结果
    index++ //下一个元素的序号
    Promise.resolve(item.value)
        .then(res => mapper(res))
        .then(res2 => {
            ret[currentIndex] = res2
            next()
        })
}

当整个迭代完后，并且元素全部执行（兑现）完，输出结果集。

var pMap = (iterable, mapper, options) => new Promise((resolve,reject) => {
    var activeCount = 0 //正在执行的元素个数
    var next = () => {
        var item = iterator.next()
        if (item.done) {  //元素全部迭代完
            if (activeCount == 0) {
                resolve(ret)  //元素全部执行（兑现）完，输出结果集
            }
            return
        }
        var currentIndex = index // 保存当前元素序号，用干存入结果
        index++ //下一个元素的序号
        activeCount++
        Promise.resolve(item.value)
            .then(res => mapper(res))
            .then(res2 => {
                ret[currentIndex] = res2
                activeCount--
                next()
            })
            .catch(err => {
                activeCount--
            })
    }
})

配置项 `stopOnError`

传入 p-map 配置项中有一个参数是 stopOnError，表示当执行遇到错误，是否终止迭代循环，所以这里在 .catch() 里面做判断；。

Promise.resolve(item.value)
    .then(res = mapper(res))
    .then(res2 => {
        // ...
    })
    .catch(err => {
        ret[currentIndex] == err // 将错误的结果也保存起来
        
        if (stopOnError) {
            hasError = true
            reject(err) // 发生错误，终止循环
        }else {
            hasError = false
            activeCount--
            next()
        }
    }

忽略错误执行结果 `pMapSkip`

mapper 函数是用户自定义的，如果 mapper 执行错误，用户期望忽略错误执行结果，只保留正确结果，这该怎么做呢？，此时 pMapSkip 就登场了；。

p-map 源码中提供了 pMapSkip，pMap5kip 是一个 Symbol 值，p-map 内部处理则是：当结果集收到的结果是 pMapSkip，则会在迭代完成后清除返回值是 pMapSkip 的元素，也就是说 mapper 处理时发生错误，用户不想要这个值，可以 reject(pMapSkip) 比如:

import pMap, { pMapSkip } from 'p-map'

var arr = [
    fetchSomething('1a', 1000, true),
    2, 3
]

var mapper = (item, index) => {
    return new Promise((resolve, reject) => {
        return item == 2 ? reject(pMapSkip): resolve(parseFloat(item)) // 元素是 2 ，抛出错误
    })
}

(async () => {
    const result = await pMap(arr, mapper, { concurrency: 2 });
    console.log(result); //=>[1,3]
})();

所以当 mapper 返回 pMapSkip 时，需要标记对应的元素。

var skipIndexArr= []

记录需要剔除的元素的位置。

var skipIndexArr = [];

Promise.resolve(item.value)
  .then((res = mapper(res)))
  .then((res2) => {
    // ...
  })
  .catch((err) => {
    if (err === pMapSkip) {
      skipIndexArr.push(currentIndex); //记录需要剔除的元素的位置
    } else {
      ret[currentIndex] == err; 

      if (stopOnError) {
        hasError = true;
        reject(err);
      } else {
        hasError = false;
        activeCount--;
        next();
      }
    }
  });

并且在迭代结束时剔除结果集中的有 pMapSkip 的元素。

if (item.done) {
  if (activeCount == 0) {
    for (var k of skipIndexArr) {
      ret.splice(k, 1);
    }
    resolve(ret);
    return;
  }
}

在数据里大的情况下，频繁使用 splice 性能可能没那么好，因为执行 splice 后，其后的元素的索引都会改变；那么就要改造下，将 skipIndexArr 改为 Map 形式.

// var skipIndexArr= []
var skipIndexArr= new Map()

记录需要删除的元素的位置。

if (err === pMapSkip) {
  skipIndexArr.set(currentIndex, err);
}

然后迭代结束时，不再在原数组里面 splice ，改为用新数组接收；push 比 splice 性能好；。

if (item.done) {
  if (activeCount == 0) {
    if (skipIndexArr.size === 0) {
      resolve(ret);
      return;
    }

    const pureRet = [];

    for (const [index, value] of ret.entries()) {
      if (skipIndexArr.get(index) === pMapSkip) {
        continue;
      }

      pureRet.push(value);
    }

    resolve(pureRet);
  }
  return;
}

在外部取消 `p-map` 的请求或者取消迭代： `AbortController`

存在某些情况，当我们不再需要 p-map 返回的结果，或者不再想要使用 p-map 时，我们就需要在外部取消 p-map 的请求或者取消迭代，这时就可以使用 AbortController；。

简单介绍下 AbortController 的用法，有一个请求 fetchSomething 。

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

想要取消 fetchSomething 请求，就需要传一个 signal 到里面；signal 是 AbortController 的实例属性；AbortController 和请求之间就是由 signal 建立关联；。

var controller = new AbortController()
var signal = controller.signal

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

fetchSomething('fetch',signal).then(res => {
    console.log('res:', res)
}).catch(err => {
    console.log('err:', err)
})

建立关联后，外部取消请求使用的是 AbortController 实例方法 controller.abort()；。

controller.abort()

然后在请求里面监听外部是否调用了 controller.abort()；有两种方式。

signal.addEventListener('abort', () => {

}, false)

或者

if (signal.aborted) {

}

完整示例:

var controller = new AbortController()
var signal = controller.signal

var fetchSomething = (str,signal) => {
    return new Promise((resolve, reject) => {
        signal.addEventListener('abort', () => {
            console.log(' addEventListener')
            reject('addEventListener取消')
        }, false)
        setTimeout(() => {
            if (signal.aborted) {
                console.log('aborted')
                reject('aborted取消')
                return
            }
            console.log('进入setTimeout')
            resolve(str)
        }, 1000)
    })
}

setTimeout(() => {
    controller.abort()
}, 500)

fetchSomething('fetch',signal).then(res => {
    console.log('res:', res)
}).catch(err => {
    console.log('err:', err)
})

500ms 后输出:

addEventListener
err: addEventListener取消
aborted

结合 p-map 使用如下:

import pMap from 'p-map';

const abortController = new AbortController();

setTimeout(() => {
	abortController.abort();
}, 500);

const mapper = async value => value;

await pMap([fetchSomething(1000), fetchSomething(1000)], mapper, {signal: abortController.signal});
// 500 ms 结束 pMap 方法，抛出错误信息.

那么 p-map 内部实现就好写了:

var pMap = (iterable, mapper, options) => new Promise((resolve, reject) => {
    var { signal } = options

    if (signal) {
        if (signal.aborted) {
            reject(signal.reason);
        }

        signal.addEventListener('abort', () => {
            reject(signal.reason);
        });
    }

    var next = () =>{
        //...
    }
})

手写的完整源码:

下面的源码也可以把 promise.then() 和 .catch() 写法改进为 async await + try catch 写法；。

let pMapSkip = Symbol('skip')

var pMap = (iterable, mapper, options) => new Promise((resolve, reject) => {
    var iterator = iterable[Symbol.iterator]()
    var index = 0 // 序号，根据可迭代对象的顺序保存结果
    var ret = []// 保存结果
    var activeCount = 0 //正在执行的元素个数
    var isIterableDone = false
    var hasError = false
    var skipIndexArr = new Map()
    var { signal, stopOnError, concurrency } = options

    if (signal) {
        if (signal.aborted) {
            reject(signal.reason);
        }

        signal.addEventListener('abort', () => {
            reject(signal.reason);
        });
    }

    var next = () => {
        var item = iterator.next()
        if (item.done) {
            isIterableDone = true
            if (activeCount == 0) {
                if (skipIndexArr.size === 0) {
                    resolve(ret);
                    return;
                }

                const pureRet = [];

                for (const [index, value] of ret.entries()) {
                    if (skipIndexArr.get(index) === pMapSkip) {
                        continue;
                    }

                    pureRet.push(value);
                }

                resolve(pureRet);
            }
            return;
        }
        var currentIndex = index // 保存当前元素序号，用干存入结果
        index++ //下一个元素的序号
        activeCount++
        Promise.resolve(item.value)
            .then(res => mapper(res))
            .then(res2 => {
                ret[currentIndex] = res2
                activeCount--
                next()
            })
            .catch(err => {
                ret[currentIndex] == err;
                if (stopOnError) {
                    hasError = true;
                    reject(err);
                } else {
                    ret[currentIndex] == err;
                    if (err === pMapSkip) {
                        skipIndexArr.set(currentIndex, err);
                    }
                    hasError = false;
                    activeCount--;
                    next();
                }

            })
    }

    for (let k = 0; k < concurrency; k++) {
        if (isIterableDone) {
            break
        }
        next()
    }
})

测试一下

1、测试 pMapSkip 。

var arr= [
    fetchSomething('1a' ,1000), // promise
    2,3, 
    fetchSomething( '4a' , 5000), // promise
    5,6,7,8,9,10
]
var mapper = (item, index) => {
    return new Promise((resolve, reject) => {
        return item == 3 ? reject(pMapSkip): resolve(parseFloat(item))
    })
}

pMap(arr, mapper, { concurrency: 2 }).then(res => {
    console.log(res)  // [1, 2, 4, 5, 6, 7, 8, 9, 10] ，剔除了 3
})

2、测试中止请求。

var controller = new AbortController()
var signal = controller.signal

pMap(arr, mapper, { concurrency: 2,signal:signal }).then(res => {
    console.log(res)
}).catch(err =>{
    console.log(err) // 500ms 后打印 AbortError: signal is aborted without reason
})

setTimeout(() =>{
    controller.abort()
},500)

3、测试 stopOnError 。

pMap(arr, mapper, { concurrency: 2,signal:signal,stopOnError:true }).then(res => {
    console.log(res)
}).catch(err =>{
    console.log(err) // Symbol(skip)
})

至此，p-map 核心功能实现完了；感兴趣的同学可以点点赞；。

最后此篇关于手写p-map（控制并发数以及迭代处理promise的库）的文章就讲到这里了,如果你想了解更多关于手写p-map（控制并发数以及迭代处理promise的库）的内容请搜索CFSDN的文章或继续浏览相关文章，希望大家以后支持我的博客！。

文章推荐： Serilog文档翻译系列（七）-应用设置、调试和诊断、开发接收器

文章推荐： glance对接ceph

文章推荐：了解final关键字在Java并发编程领域的作用吗？

文章推荐：大模型应用开发初探:基于Coze创建Agent

jquery .each 迭代
如果您有超过 1 个具有相同类名的(动态)文本框，并使用 jquery 循环遍历每个所述文本框，您是否可以假设每次选择文本框的顺序都是相同的？示例: 文本框 1 值 = 1文本框 2 值 = 2文本
Python 迭代
有人知道为什么这段代码无法顺利运行吗？它似乎不喜欢使用yield关键字进行迭代:我正在尝试从任何级别的列表或字典中挖掘所有数字(对列表特别感兴趣)。在第二次迭代中，它找到 [2,3] 但无法依次打印
PHPExcel动态单元格生成-迭代
我关于从 mysql 数据库导出数据并将其保存到 Excel 文件(多表)的创建脚本。我需要让细胞动态基因化。该脚本正确地显示了标题，但数据集为空。当我“回显”$value 变量时，我检查了数据是否存
Python 迭代？
我正在尝试在 Python 中运行模拟，由此我绘制了一个数组的随机游走图，给定了两个变量参数的设定水平。但是，我遇到了一个问题，我不确定如何迭代以便生成 250 个不同的随机数以插入公式。例如我已经
jquery .each 迭代
我是学习 jquery 的新手，所以如果这是一个相对简单的问题，我深表歉意。我有一个 ID 为 ChartstoDisplay 的 asp.net 复选框列表。我正在尝试创建 jquery 来根据是否
定义有效案例的算法/迭代
我正在尝试根据在任意数量的部分中所做的选择找出生成有效案例列表的最佳方法。也许它不是真正的算法，而只是关于如何有效迭代的建议，但对我来说这似乎是一个算法问题。如果我错了，请纠正我。实现实际上是在 Ja
scapy - DNSRR 迭代
如果我使用 sr1 为 www.google.com 发送 DNSQR，我会收到几个 DNSRR(s) 作为回复，例如(使用 ans[DNSRR].show() 完成): ###[ DNS Resou
集合字段上的 JPA 迭代
假设有这样一个实体类 @Entity public class User { ... public Collection followers; ... } 假设用户有成千上万的用户关注者。我想分页..
jquery - 如何摆脱 .each() 迭代
这个问题已经有答案了: 已关闭11 年前。 Possible Duplicate: Nested jQuery.each() - continue/break 这是我的代码: var steps =
字典上的 F# 迭代
我刚从 F# 开始，我想遍历字典，获取键和值。所以在 C# 中，我会说: IDictionary resultSet = test.GetResults; foreach (DictionaryEn
c++ - 迭代 ifstream
我知道已经有很多关于如何迭代 ifstream 的答案，但没有一个真正帮助我找到解决方案。我的问题是:我有一个包含多行数据的txt文件。 txt 文件的第一行告诉我其余数据是如何组成的。例如这是我的
javascript - 迭代 If 语句
我有 12 个情态动词。我想将每个模态的 .modal__content 高度与 viewport 高度进行比较，并且如果特定模态 .modal__content 高度 vh addClass("c
if 语句中的 javascript 迭代
在此JSFiddle (问题代码被注释掉)第一次单击空单元格会在隐藏输入中设置一个值，并将单元格的背景颜色设置为绿色。单击第二个空表格单元格会设置另一个隐藏输入的值，并将第二个单元格的背景颜色更改为红
java - 如何访问存储在单链表节点中的对象的特定变量(迭代)
这是一个非常具体的问题，我似乎找不到任何特别有帮助的内容。我有一个单链表(不是一个实现的链表，这是我能找到的全部)，其中节点存储一个 Student 对象。每个 Student 对象都有变量，尽管我在
delphi - 迭代 IHTMLElementCollection
有没有办法迭代 IHTMLElementCollection？比如 var e : IHTMLLinkElement; elementCollection:IHTMLElementCollect
java - 迭代 HashMap ？
我正在尝试用 Java 取得高分。基本上我想要一个 HashMap 来保存 double 值(因此索引从最高的 double 值开始，这样我更容易对高分进行排序)，然后第二个值将是客户端对象，如下所示
sas - 限制 %do %while 迭代
我想在宏函数中运行 while/until 循环，并限制其最大迭代次数。我找到了如何在“通常”sas 中执行此操作: data dataset; do i=1 to 10 until(con
Java - 迭代 HashMap ？
Iterator iterator = plugin.inreview.keySet().iterator(); while (iterator.hasNext()) { Player key
lambda - 序言:迭代
晚上好我有一个简单的问题，我警告你我是序言的新手。假设有三个相同大小的列表，每个列表仅包含 1、0 或 -1。我想验证对于所有 i，在三个列表的第 i 个元素中，只有一个非零。此代码针对固定的 i
recursion - 方案尾递归/迭代
我在 scheme 中构建了一个递归函数，它将在某些输入上重复给定函数 f, n 次。 (define (recursive-repeated f n) (cond ((zero? n) iden

撒哈拉

个人简介

我是一名优秀的程序员,十分优秀！

作者热门文章

滴滴打车优惠券免费领取

全站热门文章

首页

博学

6Ren·AI

商城