JavaScript 数组的高阶函数使用异步操作
吾辈的博客原文: https://blog.rxliuli.com/p/5e...
场景
吾辈是一只在飞向太阳的萤火虫
JavaScript 中的数组是一个相当泛用性的数据结构，能当数组，元组，队列，栈进行操作，更好的是 JavaScript 提供了很多原生的高阶函数，便于我们对数组整体操作。
然而，JavaScript 中的高阶函数仍有缺陷 -- 异步！当你把它们放在一起使用时，就会感觉到这种问题的所在。
例如现在，有一组 id，我们要根据 id 获取到远端服务器 id 对应的值，然后将之打印出来。那么，我们要怎么做呢？
const wait = ms => new Promise(resolve => setTimeout(resolve, ms))async function get(id) { // 这里只是为了模拟每个请求的时间可能是不定的 await wait(Math.random() * id * 100) return '内容: ' + id.toString()}const ids = [1, 2, 3, 4]你或许会下意识地写出下面的代码
ids.forEach(async id => console.log(await get(id)))事实上，控制台输出是无序的，而并非想象中的 1, 2, 3, 4 依次输出
内容: 2 内容: 3 内容: 1 内容: 4这是为什么呢？原因便是 JavaScript 中数组的高阶函数并不会等待异步函数的返回！当你在网络上搜索时，会发现很多人会说可以使用 for-of, for-in 解决这个问题。
;(async () => { for (let id of ids) { console.log(await get(id)) }})()或者，使用 Promise.all 也是一种解决方案
;(async () => { ;(await Promise.all(ids.map(get))).forEach(v => console.log(v))})()然而，第一种方式相当于丢弃了 Array 的所有高阶函数，再次重返远古 for 循环时代了。第二种则一定会执行所有的异步函数，即便你需要使用的是 find/findIndex/some/every 这些高阶函数。那么，有没有更好的解决方案呢？
思考
既然原生的 Array 不支持完善的异步操作，那么，为什么不由我们来实现一个呢？
实现思路:
- 创建异步数组类型
AsyncArray - 内置一个数组保存当前异步操作数组的值
- 实现数组的高阶函数并实现支持异步函数顺序执行
- 获取到内置的数组
class AsyncArray { constructor(...args) { this._arr = Array.from(args) this._task = [] } async forEach(fn) { const arr = this._arr for (let i = 0, len = arr.length; i < len; i++) { await fn(arr[i], i, this) } }}new AsyncArray(...ids).forEach(async id => console.log(await get(id)))打印结果确实有顺序了，看似一切很美好？
然而，当我们再实现一个 map 试一下
class AsyncArray { constructor(...args) { this._arr = Array.from(args) } async forEach(fn) { const arr = this._arr for (let i = 0, len = arr.length; i < len; i++) { await fn(arr[i], i, this) } } async map(fn) { const arr = this._arr const res = [] for (let i = 0, len = arr.length; i < len; i++) { res.push(await fn(arr[i], i, this)) } return this }}调用一下
new AsyncArray(...ids).map(get).forEach(async res => console.log(res))// 抛出错误// (intermediate value).map(...).forEach is not a function然而会有问题，实际上 map 返回的是 Promise，所以我们还必须使用 await 进行等待
;(async () => { ;(await new AsyncArray(...ids).map(get)).forEach(async res => console.log(res), )})()是不是感觉超级蠢？吾辈也是这样认为的！
链式调用加延迟执行
我们可以尝试使用链式调用加延迟执行修改这个 AsyncArray
/** * 保存高阶函数传入的异步操作 */class Action { constructor(type, args) { /** * @field 异步操作的类型 * @type {string} */ this.type = type /** * @field 异步操作的参数数组 * @type {Function} */ this.args = args }}/** * 所有的操作类型 */Action.Type = { forEach: 'forEach', map: 'map', filter: 'filter',}/** * 真正实现的异步数组 */class InnerAsyncArray { constructor(arr) { this._arr = arr } async forEach(fn) { const arr = this._arr for (let i = 0, len = arr.length; i < len; i++) { await fn(arr[i], i, this) } this._arr = [] } async map(fn) { const arr = this._arr const res = [] for (let i = 0, len = arr.length; i < len; i++) { res.push(await fn(arr[i], i, this)) } this._arr = res return this } async filter(fn) { const arr = this._arr const res = [] for (let i = 0, len = arr.length; i < len; i++) { if (await fn(arr[i], i, this)) { res.push(arr[i]) } } this._arr = res return this }}class AsyncArray { constructor(...args) { this._arr = Array.from(args) /** * @field 保存异步任务 * @type {Action[]} */ this._task = [] } forEach(fn) { this._task.push(new Action(Action.Type.forEach, [fn])) return this } map(fn) { this._task.push(new Action(Action.Type.map, [fn])) return this } filter(fn) { this._task.push(new Action(Action.Type.filter, [fn])) return this } /** * 终结整个链式操作并返回结果 */ async value() { const arr = new InnerAsyncArray(this._arr) let result for (let task of this._task) { result = await arr[task.type](...task.args) } return result }}使用一下
new AsyncArray(...ids) .filter(async i => i % 2 === 0) .map(get) .forEach(async res => console.log(res)) .value()可以看到，确实符合预期了，然而每次都要调用 value()，终归有些麻烦。
使用 then 以支持 await 自动结束
这里使用 then() 替代它以使得可以使用 await 自动计算结果
class AsyncArray { // 上面的其他内容... /** * 终结整个链式操作并返回结果 */ async then(resolve) { const arr = new InnerAsyncArray(this._arr) let result for (let task of this._task) { result = await arr[task.type](...task.args) } // 这里使用 resolve(result) 是为了兼容 await 的调用方式 resolve(result) return result }}现在，可以使用 await 结束这次链式调用了
await new AsyncArray(...ids).map(get).forEach(async res => console.log(res))突然之间，我们发现了一个问题，为什么会这么慢？一个个去进行异步操作太慢了，难道就不能一次性全部发送出去，然后有序的处理结果就好了嘛？
并发异步操作
我们可以使用 Promise.all 并发执行异步操作，然后对它们的结果进行有序地处理。
/** * 并发实现的异步数组 */class InnerAsyncArrayParallel { constructor(arr) { this._arr = arr } async _all(fn) { return Promise.all(this._arr.map(fn)) } async forEach(fn) { await this._all(fn) this._arr = [] } async map(fn) { this._arr = await this._all(fn) return this } async filter(fn) { const arr = await this._all(fn) this._arr = this._arr.filter((v, i) => arr[i]) return this }}然后修改 AsyncArray，使用 _AsyncArrayParallel 即可
class AsyncArray { // 上面的其他内容... /** * 终结整个链式操作并返回结果 */ async then(resolve) { const arr = new InnerAsyncArrayParallel(this._arr) let result = this._arr for (let task of this._task) { result = await arr[task.type](...task.args) } // 这里使用 resolve(result) 是为了兼容 await 的调用方式 if (resolve) { resolve(result) } return result }}调用方式不变。当然，由于使用 Promise.all 实现，也同样受到它的限制 -- 异步操作实际上全部执行了。
串行/并行相互转换
现在我们的 _AsyncArray 和 _AsyncArrayParallel 两个类只能二选一，所以，我们需要添加两个函数用于互相转换。
class AsyncArray { constructor(...args) { this._arr = Array.from(args) /** * @field 保存异步任务 * @type {AsyncArrayAction[]} */ this._task = [] /** * 是否并行化 */ this._parallel = false } // 其他内容... parallel() { this._parallel = true return this } serial() { this._parallel = false return this } async then() { const arr = this._parallel ? new InnerAsyncArrayParallel(this._arr) : new InnerAsyncArray(this._arr) let result = this._arr for (let task of this._task) { result = await arr[task.type](...task.args) } if (resolve) { resolve(result) } return result }}现在，我们可以在真正执行之前在任意位置对其进行转换了
await new AsyncArray(...ids) .parallel() .filter(async i => i % 2 === 0) .map(get) .forEach(async res => console.log(res))并发执行多个异步操作
然而，上面的代码有一些隐藏的问题
await之后返回值不是一个数组;(async () => { const asyncArray = new AsyncArray(...ids) console.log(await asyncArray.map(i => i * 2)) // InnerAsyncArray { _arr: [ 2, 4, 6, 8 ] }})()上面的
map,filter调用在await之后仍会影响到下面的调用;(async () => { const asyncArray = new AsyncArray(...ids) console.log(await asyncArray.map(i => i * 2)) // InnerAsyncArray { _arr: [ 2, 4, 6, 8 ] } console.log(await asyncArray) // InnerAsyncArray { _arr: [ 2, 4, 6, 8 ] }})()并发调用的顺序不能确定，会影响到内部数组，导致结果不能确定
;(async () => { const asyncArray = new AsyncArray(...ids) ;(async () => { console.log( await asyncArray.filter(async i => i % 2 === 1).map(async i => i * 2), ) // InnerAsyncArray { _arr: [ 2, 6 ] } })() ;(async () => { console.log(await asyncArray) // InnerAsyncArray { _arr: [ 2, 6 ] } })()})()
先解决第一个问题，这里只需要判断一下是否为终结操作（forEach），是的话就直接返回结果，否则继续下一次循环
class AsyncArray { // 其他内容... async then(resolve, reject) { const arr = this._parallel ? new InnerAsyncArrayParallel(this._arr) : new InnerAsyncArray(this._arr) let result = this._arr for (let task of this._task) { const temp = await arr[task.type](...task.args) if ( temp instanceof InnerAsyncArray || temp instanceof InnerAsyncArrayParallel ) { result = temp._arr } else { // 如果已经是终结操作就返回数组的值 if (resolve) { resolve(temp) } return temp } } if (resolve) { resolve(result) } return result }}现在，第一个问题简单解决
;(async () => { const asyncArray = new AsyncArray(...ids) console.log(await asyncArray.map(i => i * 2)) // [ 2, 4, 6, 8 ]})()第二、第三个问题看起来似乎是同一个问题？其实我们可以按照常规思维解决第一个问题。既然 await 之后仍然会影响到下面的调用，那就在 then 中把 _task 清空好了，修改 then 函数
class AsyncArray { // 其他内容... async then(resolve, reject) { const arr = this._parallel ? new InnerAsyncArrayParallel(this._arr) : new InnerAsyncArray(this._arr) let result = this._arr for (let task of this._task) { const temp = await arr[task.type](...task.args) if ( temp instanceof InnerAsyncArray || temp instanceof InnerAsyncArrayParallel ) { result = temp._arr } else { // 如果已经是终结操作就返回数组的值 if (resolve) { resolve(temp) } this._task = [] return temp } } if (resolve) { resolve(result) } this._task = [] return result }}现在，第一个问题解决了，但第二个问题不会解决。究其原因，还是异步事件队列的问题，虽然 async-await 能够让我们以同步的方式写异步的代码，但千万不可忘记它们本质上还是异步的！
;(async () => { await Promise.all([ (async () => { console.log( await asyncArray.filter(async i => i % 2 === 1).map(async i => i * 2), ) // [ 2, 6 ] })(), (async () => { console.log(await asyncArray) // [ 2, 6 ] })(), ]) console.log(await asyncArray) // [ 1, 2, 3, 4 ]})()可以看到，在使用 await 进行等待之后就如同预期的 _task 被清空了。然而，并发执行的没有等待的 await asyncArray 却有奇怪的问题，因为它是在 _task 清空之前执行的。
并且，这带来一个副作用: 无法缓存操作了
;(async () => { const asyncArray = new AsyncArray(...ids).map(i => i * 2) console.log(await asyncArray) // [ 2, 4, 6, 8 ] console.log(await asyncArray) // [ 1, 2, 3, 4 ]})()使用不可变数据
为了解决直接修改内部数组造成的问题，我们可以使用不可变数据解决这个问题。试想：如果我们每次操作都返回一个新的 AsyncArray，他们之间没有关联，这样又如何呢？
class AsyncArray { constructor(...args) { this._arr = Array.from(args) /** * @field 保存异步任务 * @type {Action[]} */ this._task = [] /** * 是否并行化 */ this._parallel = false } forEach(fn) { return this._addTask(Action.Type.forEach, [fn]) } map(fn) { return this._addTask(Action.Type.map, [fn]) } filter(fn) { return this._addTask(Action.Type.filter, [fn]) } parallel() { this._parallel = true return this } serial() { this._parallel = false return this } _addTask(type, args) { const result = new AsyncArray(...this._arr) result._task = [...this._task, new Action(type, args)] result._parallel = this._parallel return result } /** * 终结整个链式操作并返回结果 */ async then(resolve, reject) { const arr = this._parallel ? new InnerAsyncArrayParallel(this._arr) : new InnerAsyncArray(this._arr) let result = this._arr for (let task of this._task) { const temp = await arr[task.type](...task.args) if ( temp instanceof InnerAsyncArray || temp instanceof InnerAsyncArrayParallel ) { result = temp._arr } else { // 如果已经是终结操作就返回数组的值 if (resolve) { resolve(temp) } return temp } } if (resolve) { resolve(result) } return result }}再次测试上面的那第三个问题，发现已经一切正常了呢
- 并发调用的顺序不能确定，但不会影响内部数组了，结果是确定的
;(async () => { const asyncArray = new AsyncArray(...ids) await Promise.all([ (async () => { console.log( await asyncArray.filter(async i => i % 2 === 1).map(async i => i * 2), ) // [ 2, 6 ] })(), (async () => { console.log(await asyncArray) // [ 1, 2, 3, 4 ] })(), ]) console.log(await asyncArray) // [ 1, 2, 3, 4 ]})()- 操作可以被缓存
;(async () => { const asyncArray = new AsyncArray(...ids).map(i => i * 2) console.log(await asyncArray) // [ 2, 4, 6, 8 ] console.log(await asyncArray) // [ 2, 4, 6, 8 ]})()完整代码
下面吾辈把完整的代码贴出来
/** * 保存高阶函数传入的异步操作 */class Action { constructor(type, args) { /** * @field 异步操作的类型 * @type {string} */ this.type = type /** * @field 异步操作的参数数组 * @type {Function} */ this.args = args }}/** * 所有的操作类型 */Action.Type = { forEach: 'forEach', map: 'map', filter: 'filter',}/** * 真正实现的异步数组 */class InnerAsyncArray { constructor(arr) { this._arr = arr } async forEach(fn) { const arr = this._arr for (let i = 0, len = arr.length; i < len; i++) { await fn(arr[i], i, this) } this._arr = [] } async map(fn) { const arr = this._arr const res = [] for (let i = 0, len = arr.length; i < len; i++) { res.push(await fn(arr[i], i, this)) } this._arr = res return this } async filter(fn) { const arr = this._arr const res = [] for (let i = 0, len = arr.length; i < len; i++) { if (await fn(arr[i], i, this)) { res.push(arr[i]) } } this._arr = res return this }}class InnerAsyncArrayParallel { constructor(arr) { this._arr = arr } async _all(fn) { return Promise.all(this._arr.map(fn)) } async forEach(fn) { await this._all(fn) this._arr = [] } async map(fn) { this._arr = await this._all(fn) return this } async filter(fn) { const arr = await this._all(fn) this._arr = this._arr.filter((v, i) => arr[i]) return this }}class AsyncArray { constructor(...args) { this._arr = Array.from(args) /** * @field 保存异步任务 * @type {Action[]} */ this._task = [] /** * 是否并行化 */ this._parallel = false } forEach(fn) { return this._addTask(Action.Type.forEach, [fn]) } map(fn) { return this._addTask(Action.Type.map, [fn]) } filter(fn) { return this._addTask(Action.Type.filter, [fn]) } parallel() { this._parallel = true return this } serial() { this._parallel = false return this } _addTask(type, args) { const result = new AsyncArray(...this._arr) result._task = [...this._task, new Action(type, args)] result._parallel = this._parallel return result } /** * 终结整个链式操作并返回结果 */ async then(resolve, reject) { const arr = this._parallel ? new InnerAsyncArrayParallel(this._arr) : new InnerAsyncArray(this._arr) let result = this._arr for (let task of this._task) { const temp = await arr[task.type](...task.args) if ( temp instanceof InnerAsyncArray || temp instanceof InnerAsyncArrayParallel ) { result = temp._arr } else { // 如果已经是终结操作就返回数组的值 if (resolve) { resolve(temp) } return temp } } if (resolve) { resolve(result) } return result }}总结
那么，关于 JavaScript 中如何封装一个可以使用异步操作高阶函数的数组就先到这里了，完整的 JavaScript 异步数组请参考吾辈的 AsyncArray（使用 TypeScript 编写）。