引言问题不在不并发而在太并发假设有 1000 个文件需要上传到对象存储最直观的写法是awaitPromise.all(tasks.map((task)upload(task)));这段代码的问题不在于并发能力不足而在于并发完全不受控制。map()会立即调用 1000 次upload()只要upload()在第一个await之前已经发出请求就会瞬间制造 1000 个在途操作。远端限流、连接池上限、文件描述符上限、进程内堆积的 Buffer 与回调总有一个会先撞上。正确的做法是固定数量的执行者N 个 Runner 共享一个任务游标各自完成一个任务后领取下一个直到领完为止。这个实现只有约 40 行promise-pool.tsexportasyncfunctionmapConcurrentTInput,TResult(inputs:readonlyTInput[],concurrency:number,worker:(input:TInput,index:number)PromiseTResult,):PromiseTResult[]{if(!Number.isInteger(concurrency)||concurrency1){thrownewRangeError(concurrency must be a positive integer);}if(inputs.length0){return[];}constresultsnewArrayTResult(inputs.length);letcursor0;construnnerasync():Promisevoid{while(true){constindexcursor;cursor1;if(indexinputs.length){return;}results[index]awaitworker(inputs[index]!,index);}};construnnerCountMath.min(concurrency,inputs.length);construnnersArray.from({length:runnerCount},()runner());awaitPromise.all(runners);returnresults;}这个函数足够短可以一口气读完同时它恰好踩中了事件循环的每一个关键机制。把它彻底讲清楚所需的全部概念——闭包捕获、async 的冻结与恢复、宏任务与微任务、协作式调度——也正是理解一切 Node 并发代码所需的全部概念。本文的约定读者熟悉async/await的语法但不假定了解事件循环的内部机制。一、演示现场定时器从哪里来后文会反复出现任务睡了 5ms定时器到点这样的说法这些数字不是凭空设定的它们来自与并发池配套的演示文件promise-demo.ts。在分析机制之前先把这个演示现场交代完整。演示要模拟的场景是 1000 次文件上传。真实的上传是网络操作但分析并发机制并不需要真的联网——需要一个长相与网络 I/O 一致的替身。这个替身就是promise-pool.ts末尾的delay()exportfunctiondelay(milliseconds:number):Promisevoid{if(!Number.isFinite(milliseconds)||milliseconds0){thrownewRangeError(milliseconds must be a finite number 0);}returnnewPromisevoid((resolve){setTimeout(resolve,milliseconds);});}delay()做了一件极简单但极关键的事把定时器包装成 Promise。setTimeout(resolve, ms)注册一个ms毫秒后到点的定时器到点后宿主环境调用resolvePromise 随之落定。调用方await delay(ms)就得到了等若干毫秒后继续的效果。为什么说它是网络 I/O 的合格替身因为两者的关键性质完全一致等待发生在 JavaScript 线程之外。真实上传中请求发出去之后等待响应的工作由操作系统内核与网卡完成JS 线程在此期间是自由的定时器同样如此计时由宿主环境负责JS 线程不参与。对于主线程在任务等待期间能干什么这个问题setTimeout和真实网络请求没有区别——这正是后文一切分析的立足点。演示文件中的假上传函数asyncfunctionfakeUpload(task:UploadTask):PromiseUploadResult{// 模拟网络或磁盘等待定时器挂起期间JS 线程可以自由启动或完成其他上传awaitdelay(8(task.id%17));// 确定性失败id 为 211 的倍数时抛错保证每次运行结果可复现if(task.id0task.id%2110){thrownewError(simulated network failure for${task.fileName});}return{id:task.id,objectKey:uploads/${task.id}-${task.fileName},};}两处设计都值得注意。其一延迟取8 (task.id % 17)毫秒——每个任务的等待时长在 8~24ms 之间变化且各不相同这是刻意为之如果所有任务等待时长相同它们会同时到点、按固定顺序完成看不出谁先到点谁先继续的动态调度错开的时长才能让完成顺序真实地乱起来。其二失败是确定性的id 为 211、422、633、844 的 4 个任务这样演示的每一次运行都能得到完全相同的结果便于验证。演示的主流程只有三行consttasks:UploadTask[]Array.from({length:1_000},(_,id)({id,fileName:file-${id}.bin,}));constoutcomesawaitmapConcurrentSettled(tasks,32,fakeUpload);1000 个任务32 个 Runner实际运行输出submitted: 1000 succeeded: 996 failed: 4failed: 4正好对应 4 个确定性失败机制与预期一致。mapConcurrentSettled与mapConcurrent的区别在第五节展开此处只需知道它不让单个失败拖垮整批任务。至此演示现场已经完整fakeUpload提供任务delay()提供等待setTimeout提供到点通知。后文分析mapConcurrent时出现的每一个定时器指的都是这条链。二、内存格局什么被共享什么是私有要理解这段代码首先要回答一个问题32 个 Runner 同时操作cursor和results这些数据究竟住在哪里堆Heap results 数组对象 ← 所有 Runner 激活共享同一份 context 盒子 { cursor } ← 所有 Runner 激活共享同一份 每次调用 runner() 产生的一次激活 自己的 index、自己的循环进度 ← 私有一次调用一份results是数组数组是对象对象一律分配在堆上这没有例外。cursor的情况更有意思。它只是一个let声明的数字如果没有被内层函数引用V8 会把它放在mapConcurrent的栈帧里函数返回即销毁。但runner这个闭包引用了它V8 对被闭包捕获的变量的处理方式是把它从栈帧挪进一个堆上的 context 对象可以将其理解为一个隐形的{ cursor: 0 }盒子栈帧中只保留指向盒子的指针。这正是闭包捕获的变量其生命周期脱离函数调用在这条代码里的具体体现。需要严格区分的是runner这个函数只有一份一个闭包但它被调用了 32 次产生 32 个相互独立的激活activation。每个激活各自记录自己的index与循环进度而它们读写的cursor与results是堆上的同一份。共享的准确含义不是 32 份拷贝而是 32 个激活指向同一个盒子。三、启动阶段没有创建任何线程construnnersArray.from({length:runnerCount},()runner());这行代码没有创建任何线程。从头到尾只有一根主线程。那么它做了什么runner是 async 函数而 async 函数的调用规则是调用它它会同步地向下执行直到撞上第一个await在await处它冻结自身把当前进度打包存好然后立即返回一个 Pending 状态的 Promise 给调用方。因此Array.from的回调执行 32 次每次都经历相同的过程调用 runner() → 同步执行index cursor例如 0cursor 变为 1检查边界 → 调用 worker(inputs[0], 0) 在演示中即 fakeUpload同步执行到 await delay(...) → delay() 内部 setTimeout(resolve, ...) 注册定时器返回 Pending Promise → runner 在 await 处冻结把 { index: 0, 循环位置 } 存入堆上的续体对象 → runner() 返回一个 Pending Promise注意定时器在这条链中的位置它是在每个 Runner 的同步执行段内被注册的。32 次runner()调用是一口气同步完成的主线程在此期间没有被任何人打断。即使某个任务的定时器时长为 0ms它的回调也无法插入这个同步窗口——主线程不让出事件循环中排队的任何回调都只能等待。由此得到一个可验证的推论前 32 个任务的领取顺序是确定的 0、1、2……31与之对应的 32 个定时器也在这个窗口内全部注册完毕。只有启动阶段结束之后领取顺序才交给任务完成的先后决定。启动完成的瞬间现场如下runners 数组32 个 Pending Promise对应 32 个被冻结的激活 堆上32 个续体各自记着 index 0..31 宿主环境libuv32 个定时器已在计时 主线程执行 await Promise.all(runners)mapConcurrent 自身也冻结线程彻底空出while循环不是并发原语。它只是一个尚未执行到头的普通循环每个激活各自停在循环中await那一行。真正制造并发的是冻结 稍后恢复的机制而不是循环更不是线程。四、宏任务、微任务、回调与续体激活被冻结之后靠什么唤醒答案藏在事件循环的两级队列里而第一节的delay()正是理解这两级队列的入口。事件循环处理工作的队列有两条优先级不同宏任务macrotask事件循环每一圈只取出一个来执行。setTimeout到点后的回调、I/O 完成回调、事件回调都以宏任务的形式排队。微任务microtask每执行完一个宏任务或一段同步脚本引擎会立刻清空整个微任务队列——包括清空过程中新排进来的微任务——然后才允许进入下一个宏任务。promise.then(...)中的函数、queueMicrotask的入队项都是微任务。与此相关的两个术语回调callback你亲手交给宿主环境的函数含义是到时候替我执行。在delay()里setTimeout(resolve, ms)中的resolve就是回调——它是宏任务的载体。续体continuationawait把 async 函数剩下的那半段代码打包出来的对象。它不是你显式写出的函数而是引擎从函数现场切出来的下半截以微任务的形式排队。await p在语义上约等于p.then(下半截代码)。在演示中Runner 的续体就是写回 results、回到循环顶部、领取下一个任务这半段。把整条唤醒链串起来定时器到点第一节由宿主环境计时JS 线程不参与 → 事件循环取出一个宏任务回调 resolve 执行 → resolve 让 delay 的 Promise 落定把 Runner 的续体排入微任务队列 → 当前宏任务结束引擎立即清空微任务队列 → 续体执行激活从冻结的 await 那一行继续就像函数从未离开过所以微任务永远在相邻两个宏任务之间插队执行。await delay(...)之后的代码能够很快继续靠的正是微任务的这种高优先级。五、严格时间线3 个 Runner7 个任务现在用一个可完整推演的实例把前几节的内容落到一条时间线上。设定concurrency 3任务共 7 个index 0~6。有一点需要先说明演示程序中相邻任务的延迟只差 1ms8 (id % 17)画在时间线上所有唤醒几乎同时发生反而看不清机制。因此下图把各任务的等待时长拉开任务 0 睡 30ms、任务 1 睡 5ms、任务 2 睡 12ms后续任务各睡 10ms机制与演示程序完全一致只是时间比例经过了夸张处理便于观察谁先到点谁先继续。图中纵轴是四条泳道R1、R2、R3 三个激活的生命线以及最下方一条主线程占用条。蓝色实块表示占用主线程的同步执行段蓝色虚线表示激活在睡觉不占线程定时由宿主环境负责。逐个时刻推演t0 R1 启动领 index0cursor→1注册 30ms 定时器冻结 t1 R2 启动领 index1cursor→2注册 5ms 定时器冻结 t2 R3 启动领 index2cursor→3注册 12ms 定时器冻结 t3 启动窗口关闭。Promise.all 挂起主线程空闲事件循环开始工作 t0~t2 是三个先后执行、零缝隙的时刻不是同一时刻 同步窗口内宏任务与微任务都无法插入 t4 R2 的 5ms 先到点宏任务回调 resolve → 微任务续体 写回 results[1] → 领 index3cursor→4 → 注册 10ms 定时器 → 冻结 t5 R3 的 12ms 到点写回 results[2] → 领 index4cursor→5 → 睡 10ms t6 R2 再到点15ms写回 results[3] → 领 index5cursor→6 → 睡 10ms t7 R3 再到点22ms写回 results[4] → 领 index6cursor→7 → 睡 10ms t8 R2 再到点25ms写回 results[5] → 领号 7越界 → return t9 R1 的 30ms 终于到点写回 results[0] → 领号越界 → return t10 R3 再到点32ms写回 results[6] → 领号越界 → return t11 Promise.all 的三个 Promise 全部落定 → mapConcurrent 返回 results这张图最重要的纪律画在最下面主线程占用条只有一格宽。任意时刻主线程只执行一段续体所有蓝块在时间轴上必然首尾相接永不重叠。三条激活泳道只是谁还活着、在等什么的记录不是三条并行轨道。从这张图上可以直接读出两个最初看似需要证明的结论cursor 为什么不需要锁。恢复执行的唯一入口是微任务微任务只在主线程空闲时逐个执行一段续体跑完到下一个await之前其他续体不可能插入。index cursor; cursor 1两行之间没有await因此这两行是原子的。这套机制的安全性完全建立在切换只发生在 await 点之上——这是协作式调度不是抢占式调度。假如领取与自增之间隔着await就会出现两个 Runner 领到同一个号的事故。结果为什么不乱序。代码中没有任何地方使用push。每个激活从领号那一刻起就私有地记住了自己的index冻结时存入续体恢复时原样取出唤醒后第一件事就是把结果写回results[index]。完成顺序是乱的写入位置永远不乱。六、三个不需要新代码就能推出的结论严格时间线的价值在于它可以当作推理工具使用。以下三个结论不需要阅读任何新代码直接从机制中推导即可。推论一如果 worker 是纯同步 CPU 函数这个池子会退化。假设worker没有任何真实 I/O返回一个已经 resolved 的 Promise。时间线的形状几乎不变——每个await依然让出续体排入微任务队列先进先出领号依然轮流、不会错乱。但每个蓝块变得很长纯计算块与块之间主线程没有空隙处理任何其他事情总耗时等于所有任务耗时之和定时器与 I/O 回调全部被堵在块外。并发池重叠的是等待CPU 任务没有等待可重叠。这就是I/O 用并发池、CPU 用多进程在时间线上的样子图没变但每一格都变成了独占。推论二Promise.all拒绝之后其他 Runner 仍在继续执行。某个 Runner 的worker抛错其 Promise 拒绝Promise.all立即拒绝mapConcurrent抛错返回。但其余激活的续体依然挂在各自的定时器上到点照常唤醒、照常领号执行——没有任何机制通知它们停止只是它们的结果再无人接收。如果需要真正的取消中断网络请求、关闭文件必须把AbortSignal一路传递到最底层的 APIPromise 本身没有强制取消的能力。推论三失败收集应该发生在单任务层面。上述问题有一个干净的解法也正是演示程序实际使用的mapConcurrentSettled()exportfunctionmapConcurrentSettledTInput,TResult(inputs:readonlyTInput[],concurrency:number,worker:(input:TInput,index:number)PromiseTResult,):PromiseArrayTaskOutcomeTResult{returnmapConcurrent(inputs,concurrency,async(input,index){try{return{status:fulfilled,value:awaitworker(input,index)};}catch(reason:unknown){return{status:rejected,reason};}});}在每个任务外面包一层try/catch把错误转成普通结果{ status: rejected, reason }返回。Runner 自身永远不会拒绝整批任务必然执行完毕调用方再逐项检查状态。演示中 4 个确定性失败没有拖垮其余 996 个任务靠的就是这层包装。它不是新机制只是对续体何时拒绝这一既定行为的合理利用。七、边界与引申这个 40 行函数的正确性依赖两个前提它们值得在结尾处明确指出。第一单线程前提。“cursor 不需要锁成立是因为所有激活共享一根主线程切换只发生在await点。一旦把同样的共享状态搬到多线程环境例如 Worker Thread 配合SharedArrayBuffer领取与自增之间就可能被真正抢占届时需要Atomics、锁或无锁算法。同一段代码在两种环境下的安全性截然不同分界线就是是否存在抢占”。第二I/O 前提。池子的收益来自重叠等待而等待发生在线程外正是第一节用delay()模拟的那个性质。对于 CPU 密集型任务正确的方向是多个进程或线程占据多个核心让蓝块在时间轴上真正重叠——那是另一套机制子进程池、IPC、任务协议也是另一个话题。回到开头的问题1000 个上传任务concurrency 32的mapConcurrent足以胜任。它只有 40 行但它背后的每一行——堆上的共享盒子、冻结的激活、两级队列、一格宽的主线程占用条——都是事件循环这一整套机制的直接投射。读懂这个函数事件循环就不再是一个需要背下来的概念而是一种可以随手推演的行为。附本文涉及的两个源文件为 promise-pool.ts并发池与 delay 实现和 promise-demo.ts1000 次模拟上传演示。时间线图由 3 Runner / 7 任务的简化模型绘制所有时刻可换算为毫秒验证自洽t45ms、t512ms、t615ms、t722ms、t825ms、t930ms、t1032ms。
