先记住这个答案
并发池里同时跑着好几个任务,完成的先后取决于每个任务的耗时,和发起顺序无关。所以不能在任务完成时 push 结果,要在发起时就记下它的下标 i,完成后写到 results[i]。结果数组提前按任务总数建好,每个位置只属于一个任务,谁先回来都不会挤占别人的位置。这其实就是 Promise.all 保证结果顺序的方式。
- 按下标写
results[i],不要push - 下标要在
await之前取,存在局部变量里 - 结果数组提前按任务数建好,长度固定
- 失败也要占住自己的位置,否则后面的对不上
按下标写回的并发池
先看一个限制同时只跑 2 个任务的池子。三个任务耗时分别是 30、10、5 毫秒,完成顺序和发起顺序完全不同。
const task = (name, ms) => () =>
new Promise((resolve) => setTimeout(() => { console.log('完成', name); resolve(name) }, ms))
async function pool(tasks, limit) {
const results = new Array(tasks.length) // 长度提前定好
let next = 0
async function worker() {
while (next < tasks.length) {
const i = next++ // 在 await 之前占住下标
results[i] = await tasks[i]() // 按下标写回,不用 push
}
}
await Promise.all(Array.from({ length: limit }, worker))
return results
}
pool([task('A', 30), task('B', 10), task('C', 5)], 2).then((r) => console.log(JSON.stringify(r)))查看输出与解释
完成 B
完成 C
完成 A
["A","B","C"]两个 worker 一开始分别领走 A 和 B。B 在第 10 毫秒完成,它所在的 worker 接着领走 C,C 在第 15 毫秒完成,A 最后在第 30 毫秒完成。如果用 push,结果会是 ["B","C","A"]。按下标写回后,顺序和任务列表一致。
为什么下标一定要在 await 之前取
next 是几个 worker 共享的计数器。JavaScript 是单线程的,const i = next++ 这一行执行时不会被打断,所以每个 worker 拿到的下标一定不同。
但只要中间隔了一个 await,情况就变了。假如写成先 await tasks[next](),完成后再 results[next] = ... 并让 next 加一,等这个任务回来时,别的 worker 可能已经把 next 往后推了好几格,结果就写到了别人的位置上。所以原则是:共享状态在 await 之前读出来,存进这次任务自己的局部变量。
用类比来说,下标就像排队取的号码牌。取号那一刻号码就定了,什么时候轮到你办完业务不影响你的号码。
另外两种需求:按完成顺序处理、按顺序分段输出
保持发起顺序不是唯一的需求,面试里经常会接着问变体。
- 谁先完成先处理
- 比如批量上传图片,希望每张传完就立刻在界面上打勾,这时要的正是完成顺序。在任务完成时直接调用回调,或者
push到一个数组里,就是按完成顺序。两种需求可以同时满足:完成时既写results[i],又调用一次onProgress(i, value)。 - 按原顺序,但前面的好了就先输出
- 比如分段加载一篇长文,第 3 段先回来也不能先显示,要等第 1、2 段。做法是维护一个指针
cursor,指向下一个该输出的下标。每有任务完成,就从cursor开始往后检查,只要results[cursor]已经有值就输出并把cursor加一,遇到空位就停下,等后面的任务完成再继续。
容易答错的地方
- 用
results.length判断全部完成 results[5] = x会让数组长度直接变成 6,哪怕前面 5 个位置还空着。判断是否全部完成要用单独的计数器,或者像上面那样等所有 worker 结束。- 某个任务失败,整个池子就断了
- 上面的写法里,一个任务拒绝会让它所在的 worker 抛错,
Promise.all随之拒绝,但其他 worker 还在继续跑。如果希望失败也保留位置,就在 worker 里try/catch,写成results[i] = { status: 'rejected', reason: err },效果和Promise.allSettled一样。 - 以为
Promise.all会按顺序执行任务 Promise.all只保证结果按传入顺序排列,不控制执行。传给它的 Promise 在创建时就已经开始跑了,它既不会串行,也不会限制并发数量,所以才需要自己写池子。
面试官还会怎么问?
为什么要先 new Array(tasks.length),直接用空数组不行吗?
用空数组 [] 也能按下标写,结果是一样的。区别在过程中:[] 的长度会随写入的最大下标跳动,new Array(tasks.length) 一开始长度就等于任务数,没完成的位置是空位。长度固定后,意图更清楚,调试时也能一眼看出哪些位置还没回来。
任务列表是异步生成器、总数未知时怎么办?
下标改成在领任务时递增生成,不依赖总数。结果可以先存进一个 Map,key 是下标,全部结束后再按下标排序输出。
如果要求任意一个失败立刻停掉其他任务呢?
给池子加一个共享的 AbortController,任务函数接收 signal。任何任务失败时调用 abort(),正在跑的请求被取消,worker 的循环里也检查 signal.aborted,不再领新任务。
参考资料
示例用于理解所注明的运行环境与边界;延伸学习可结合原文中的更多案例。