# 异步、并发与取消 ## 本章解决什么问题 一个 AI 知识库要把三百段文本送去生成向量。逐条 await 太慢，一次 Promise.all 又可能触发服务限流、撑满连接池，并在用户取消后继续花费调用额度。本章目标是实现一个有并发上限、保留输入顺序、支持协作取消的任务池，清楚区分等待、并发、并行与重试。前置是上一章的进程模型，以及 Promise 的兑现和拒绝语义。 你已经在 Vue 搜索框里处理过“先发出的请求后返回”问题。后端面对相同的时间顺序问题，还要多考虑成本、共享资源以及调用方断开后的剩余工作。前端丢弃旧结果只影响显示，后端忽略结果却不停止调用，模型供应商仍可能继续计算并计费。 ## 先把四个概念分开 串行是前一个完成才开始后一个；并发是同一段时间存在多个未完成任务；并行是多个执行单元真正同时工作。Node 的网络并发不要求每个请求有一条 JavaScript 线程。限制并发数的意思是限制“正在处理”的任务数，而不是把所有 Promise 创建完以后再分批等待；Promise 往往在创建时已经开始执行，因此限流器应接收函数或数据，由工作者决定何时启动。 `Promise.all` 保留输入顺序，但其中一个任务拒绝时，只会让聚合 Promise 尽快拒绝，不会撤销已经启动的其他工作。`Promise.allSettled` 可以等待所有结果，适合收集和清理，但它自身同样没有并发上限。失败后需不需要停止其他任务，是产品合同：批量导入可允许逐条失败，扣减同一份预算可能要求尽快停止。本章选择一项失败后取消剩余可取消任务，并在所有工作者收尾后抛出错误。 超时也不等于取消。把原请求与一个超时 Promise 放进 race，只改变谁先向调用方返回，落后的请求可能仍在运行。可靠的实现要把 AbortSignal 传到真正执行 I/O 的 API。已经提交到外部系统的操作可能无法撤销，所以超时之后还需要依靠幂等键或状态查询判断真实结果。 ## 核心 API 合同 `new AbortController()` 创建控制器；`controller.signal` 是只读观察接口；`controller.abort(reason)` 发出一次停止通知。`signal.aborted` 表示状态，`signal.throwIfAborted()` 在已经取消时抛出 reason。取消不是强制抢占，任务函数必须接收信号并在适当位置响应。 `AbortSignal.timeout(ms)` 创建超时信号，`AbortSignal.any(signals)` 合并信号，任意一个取消都会触发组合信号。本章要求 Node 22.22 或 24，避免读者在旧版本里遇到接口缺失。[Node 全局 API](https://nodejs.org/docs/latest-v24.x/api/globals.html) 定义了这些方法。`node:timers/promises` 的定时器支持 signal，取消时通常拒绝为 AbortError；不同 API 的错误形状并非完全一致，不能只凭一个字符串推断全部失败原因。 我们设计的 `mapLimit(items, limit, mapper, { signal })` 返回与 items 等长、同顺序的结果数组。limit 必须为正整数；mapper 收到当前项、原下标与组合信号；一项失败会取消同批任务，并在收尾后拒绝。这个合同没有实现每秒请求数限制：并发二且每项只需一毫秒，仍可能产生很高的请求频率。 ## 完整示例：观察并发上限与超时 环境：Node 22.22 或 24；无依赖。保存为 `concurrency.mjs`，执行 `node concurrency.mjs`。模拟任务仅使用本地定时器，不消耗模型额度。 ```js concurrency.mjs // concurrency.mjs import assert from 'node:assert/strict'; import { setTimeout as delay } from 'node:timers/promises'; async function mapLimit(items, limit, mapper, { signal } = {}) { if (!Number.isInteger(limit) || limit < 1) { throw new RangeError('limit 必须是正整数'); } const local = new AbortController(); const combined = signal ? AbortSignal.any([signal, local.signal]) : local.signal; const results = new Array(items.length); let cursor = 0; let firstError; let failed = false; async function worker() { try { while (true) { combined.throwIfAborted(); // 取号前没有 await，同一事件循环内不会领取同一下标。 const index = cursor; cursor += 1; if (index >= items.length) return; results[index] = await mapper(items[index], index, combined); } } catch (error) { if (!failed) { failed = true; firstError = error; local.abort(error); } } } const workers = Array.from( { length: Math.min(limit, items.length) }, () => worker(), ); // 等待已启动的工作者释放资源，再把错误交给上层。 await Promise.allSettled(workers); if (failed) throw firstError; combined.throwIfAborted(); return results; } let active = 0; let peak = 0; const durations = [80, 20, 40, 10]; const values = await mapLimit(durations, 2, async (ms, index, signal) => { active += 1; peak = Math.max(peak, active); try { await delay(ms, undefined, { signal }); return `结果${index + 1}`; } finally { // 成功、失败和取消都必须归还并发名额所代表的资源。 active -= 1; } }); assert.deepEqual(values, ['结果1', '结果2', '结果3', '结果4']); assert.equal(peak, 2); assert.equal(active, 0); console.log('结果顺序正确，并发峰值为 2，活动任务归零'); try { await mapLimit([100, 100, 100], 2, async (ms, index, signal) => { await delay(ms, undefined, { signal }); return index; }, { signal: AbortSignal.timeout(30) }); assert.fail('超时批次不应成功'); } catch (error) { assert.ok(['AbortError', 'TimeoutError'].includes(error.name)); console.log('超时已传播，工作者已收尾'); } ``` 预期固定输出是两条成功说明，进程以零退出。耗时只用于制造不同完成顺序，不把精确毫秒数当成验收标准，因为调度、机器负载和虚拟化环境都会影响时钟。遇到断言错误时，Node 会输出失败位置并返回非零状态。 ## 逐段解释为什么这样写 cursor 的读取与自增之间没有 await，工作者领取任务的步骤在当前 JavaScript 执行片段内完成。任务一旦 await，就把执行机会让给其他工作者；因此最多有 limit 个 mapper 尚未完成。结果写到原下标，避免“完成顺序”变成“输入顺序”。前端在处理搜索建议时也应明确这两种顺序，不能靠网络返回时机碰巧一致。 错误分支记录第一个观测到的错误，并通过本地控制器通知同批任务停止。额外的 failed 布尔值是为了允许某些代码抛出 undefined 等非标准值；真实项目仍应统一抛出 Error。等待 allSettled 后再返回，让调用方看到失败时资源已尽可能收尾。若 mapper 完全不响应 signal，这个等待仍可能拖很久，证明取消合同需要全链路配合。 这份任务池适合有限数组。对于持续到达的任务流，还需明确排队容量、拒绝策略、公平性和关机方式。给一个无限数组设置并发二，并不能限制已经排队的数据占用。AI 应用中可同时设置每租户配额和全局容量，防止某个用户的文档导入挤占所有交互式聊天请求。 ## 边界与常见错误 不要用 `forEach(async () => ...)` 期待外层等待完成，它不会收集回调返回的 Promise。不要在 catch 中无限重试；限流、网络瞬断可能适合带抖动退避，参数错误、鉴权失败通常应立即结束。每次重试需要共享总截止时间与取消信号，否则三次各十秒的重试会突破接口承诺的十秒响应时间。 HTTP 客户端断开不必然等于后台任务应该停止。即时预览可以取消，已经确认提交的导入作业则应独立持续运行并通过任务状态查询。这属于接口语义，应先决定再连接 signal。数据库事务内也不要启动许多与事务无关的并发请求，长时间占有连接和锁会扩大失败影响。 ## 从页面请求竞争走向服务端资源调度 在“团队文档与任务助手”中，一份文档可能拆成一百个片段，同时还存在交互式聊天、文件解析和摘要生成。前端关注哪个请求结果应该显示，服务端还要关心这些请求共同占用多少数据库连接、内存和供应商额度。如果只为每个接口单独写 Promise.all，全站仍可能在多个用户同时操作时超过共享资源容量。 并发上限因此必须有作用范围。某个函数内最多四项，不等于整个进程最多四项；十个请求各启动四项，会得到四十项。进程上限也不等于所有实例的上限，扩容以后总并发会增加。实际产品可以对单文档、单租户与供应商账户分别设置限制，让一个大文件不会压住所有人的即时问答。 排队同样消耗资源。限制执行中的任务却接受无限请求，可能把压力从外部接口转移到本地内存。需要明确队列容量，超过容量时是返回忙碌、延迟提交还是写入持久任务队列。交互式请求通常不适合等待几十分钟前面的大批量任务；可以区分前台与后台任务通道，让容量规则符合用户期待。 ## API 细读：Promise 聚合与取消对象 `Promise.all(iterable)` 接受可迭代对象，返回一个新的 Promise；空输入兑现为空数组，结果按输入顺序排列，任一输入拒绝会让聚合拒绝。它不提供 limit 参数，也不会调用剩余任务的取消方法。输入若已经是启动后的 Promise，调用 all 只是观察和聚合，不是在这一刻统一启动任务。理解执行开始时间，比记住函数名称更重要。 `Promise.allSettled(iterable)` 返回每个输入的 status，以及 value 或 reason。它适合批量预览这种需要逐项结果的合同，但如果其中一个 Promise 永远不结束，聚合也会一直等待。`Promise.race(iterable)` 只采用最先完成者的结果，空输入会一直保持等待；落后任务既没有被撤销，也没有被自动清理。不能把 race 当成通用资源管理工具。 AbortController 没有必填构造参数；signal 初始未取消，abort 可选传 reason，第一次取消后状态不会恢复。一个控制器不能取消后再“重置复用”，新一轮任务应创建新控制器。监听 abort 事件时应考虑一次性监听和及时移除，避免长期对象积累监听器。调用任务之前先检查已经取消的 signal，可以避免无意义地启动新的网络请求。 `AbortSignal.timeout(milliseconds)` 接收非负等待值，生成会超时取消的信号；`AbortSignal.any(signals)` 合并多个来源，并保留最先触发的取消原因。超时信号表达的是本地等待预算，不是对外部系统的事务回滚。自定义 mapper 应约定 signal 参数位置、是否会取消底层操作，以及取消后何时完成资源释放，否则调用方无法正确推理生命周期。 ## 最小实验：Promise.race 没有停止落后任务 保存为 `race-timeout.mjs`，Node 22.22 或 24 执行 `node race-timeout.mjs`，无依赖。实验只修改本地计数器，不发网络请求；它模拟客户端已经收到超时，但后台副作用仍然发生。 ```js race-timeout.mjs // race-timeout.mjs import assert from 'node:assert/strict'; import { setTimeout as delay } from 'node:timers/promises'; let effects=0; async function uncooperative() { await delay(60); effects+=1; return '完成'; } const work=uncooperative(); await assert.rejects(()=>Promise.race([ work, delay(10).then(()=>{throw new Error('等待超时');}), ]),/等待超时/); await work; assert.equal(effects,1); console.log('race 已超时，但原任务仍产生一次副作用'); effects=0; const signal=AbortSignal.timeout(10); await assert.rejects(async()=>{ await delay(60,undefined,{signal}); effects+=1; },{name:'AbortError'}); assert.equal(effects,0); console.log('底层等待接受 signal，取消后没有执行后续副作用'); ``` 预期输出两条说明。第二段之所以没有副作用，是因为示例把副作用安排在一个可取消等待之后；如果外部系统在取消前已经提交，结果仍可能发生。取消能阻止尚未开始的步骤、终止支持取消的等待，但不能把已经完成的世界倒回去。需要同时具备幂等与状态查询，才能处理模型任务或扣费请求的结果不确定。 ## 截止时间、重试与供应商限流 假设接口总预算十秒，第一次调用耗时八秒后失败。如果重试再次分配完整十秒，最终响应可能远超接口承诺。更清楚的做法是在入口确定总截止时间，每次尝试只使用剩余预算，并在排队、退避和请求阶段共享取消意图。排队时间同样属于用户等待，不能从总耗时统计中消失。 并发数限制与每秒请求数限制是不同控制。四个并发任务每个十毫秒完成，仍可能快速发送很多请求；供应商同时限制并发与每分钟额度时，要分别满足。服务返回限流信号后可以按其明确合同选择等待时间，但不应在多个实例中同时立即重试，形成同步洪峰。退避加入抖动可以分散重试时刻，仍需设置最大尝试次数和总预算。 重试的依据是失败类别与操作语义。网络暂时断开、容量不足可能可以重试；参数错误、无权访问、已删除文档通常没有重试价值。创建任务或扣费这样的副作用操作，即使错误看起来是网络问题，也必须沿用幂等键。不要把一个通用 retry 包装器无区别套在所有函数上，更不能无限重试来掩盖错误。 ## 原练习的完整参考：逐项失败与批次取消并存 保存为 `settled-limit.mjs`，无依赖，执行 `node settled-limit.mjs`。合同是业务失败保留在原下标，用户取消使整个批次拒绝，所有已经启动的工作者先完成收尾。错误结果只包含公开 code 与 message，不包含堆栈或凭据。 完整代码已收录在本章末尾的练习参考答案中；可先阅读说明，再展开复制运行。 这份实现把业务错误变成正常结果，因此工作者还能继续领取后面的片段。取消检查出现在领取前、mapper 完成后与 catch 中，是为了避免取消恰好发生在结果返回附近时被误当成功。检查并不能抢占不合作的计算；如果 mapper 一直同步占用线程或从不结束，外层仍无法及时完成。接口合同应明确这种限制，而不是用“支持 AbortSignal”四个字替代全链路验证。 ## 结果顺序、完成顺序与进度更新 批量结果通常按输入顺序返回，用户才知道第七个错误对应哪段文档；进度却按完成顺序增长，可以先看到快任务完成。两者可以同时存在：每个任务带稳定 index 或业务 id，完成时发出进度事件，最终数组按原位置组装。不要拿“已完成五项”推断前五项都完成了，也不要用完成计数替代具体失败列表。 批量任务中的百分比也需要谨慎。十个片段可能长度差异很大，完成五个不一定代表完成一半成本。最初可以提供已完成数量与总数量，明确它是任务计数；若需要更准确估计，再引入字节数、token 数或历史耗时权重。后端不应为了让进度条平滑而伪造完成事实。 前端取消预览时，通常希望停止无用工作；用户确认提交的文档入库任务，则可能希望关闭页面后继续。入口应把这两种意图区分为临时请求与持久任务。临时请求的取消信号可绑定连接生命周期，持久任务则有独立取消 API 和权限检查，不依赖某个浏览器连接是否还活着。 ## 调试时先观察排队与活动任务 在任务池中记录 pending、active、completed、failed 和 cancelled 数量，峰值比单次最终值更有用。资源泄漏常在失败与取消路径出现：成功十次都正常，一次超时后 active 永远不回零。把资源计数增加放在实际领取之后，减少放在 finally，才能准确观察所有退出路径。 日志应携带批次编号、任务编号、尝试次数和停止原因。仅记录“请求失败”无法区分用户取消、截止时间、供应商拒绝和解析异常。取消是预期控制流时，可以采用与未知故障不同的日志级别，但不要把它统一吞成空数组；调用方需要知道结果是不完整还是确实没有数据。 测量吞吐时同时观察失败率与尾部等待。把并发从四调到四十，平均耗时可能暂时下降，但限流重试增加，最慢用户反而等待更久。应在代表性文档长度和租户混合负载下比较完成时间、外部错误、内存与连接使用，再选择容量。并发参数不是越大越先进，而是对共享资源预算的承诺。 ## 集成时让取消与资源所有权一起传递 业务服务接收 signal 后，应把它继续传到支持取消的 HTTP、定时器或文件操作；不支持的库需要明确超时和清理策略，不能只在最外层 race。数据库查询取消有驱动和协议层细节，不要假设所有库都接受通用 signal 参数；先查实际驱动合同，再实现可验证的取消路径。 跨服务传播的是截止时间与任务身份等协议数据，不是把 JavaScript AbortSignal 对象序列化发给另一台机器。下游需要把这些数据重新构造成自己的预算，并在返回时说明结果状态。对于可计费模型任务，取消后仍应收集最终可获知的用量，防止因为页面不再等待而丢掉成本记录。 当任务数量大到不能一次装进数组时，应使用异步迭代器或持久队列逐步读取；每次只保留正在处理与必要结果。输出本身也可能很大，最终 Promise 返回一个百万项数组仍会占用内存。任务池只是调度层，输入流、输出存储、错误保留和进度传输都需要单独设计。 ## 多个调用方共享任务时，谁有权取消 为了节省成本，两个页面可能共享同一份摘要计算。此时其中一个用户关闭面板，并不意味着另一个用户也不需要结果。如果直接把第一个调用者的 signal 作为共享任务的总开关，它取消时会误伤其他等待者。应区分“某个等待者不再等待”与“底层工作不再需要”：等待者可以独立结束，底层任务可在没有任何需求方以后再按策略取消，或者继续完成并回填缓存。 这与单次请求独占资源的情况不同。引用计数、任务拥有者和结果缓存会增加实现复杂度，因此不要在最开始就让所有调用都共享一个全局 Promise。先确定资源是否真正共享、共享结果是否跨权限安全、失败是否应传播给全部等待者，再决定是否合并请求。共享错误也应能恢复，一次失败后必须允许后来的请求重新启动。 ## 公平性会影响用户体验 一个团队一次导入一万段文本，如果队列严格按到达顺序处理，后来另一个团队提交的十段可能等待很久。即使总并发始终符合上限，体验仍然不公平。可以按租户分队列轮流领取、为交互式任务保留部分容量，或限制每个批次一次可占用的名额。不同策略会改变吞吐与等待分布，应根据产品的付费等级和服务承诺显式决定。 公平调度不意味着所有任务都必须同时启动，也不能突破供应商总配额。它是在相同总预算内决定先服务谁。监控时除了全局队列深度，还应观察各租户的最老等待时间；否则全站平均数据正常，个别用户却可能一直没有进展。 ## 练习：允许部分失败 把本例改成批量预览接口：某一项失败不取消其他项，每个位置返回 `{ ok: true, value }` 或 `{ ok: false, error }`，但用户主动取消仍应停止整个批次。提示：在 mapper 的包装层捕获业务错误，遇到 signal 已取消则重新抛出；错误输出保留稳定代码，不向用户返回内部堆栈。
参考答案（含完整可运行实现） 把传给 mapLimit 的函数包装成 try/catch。正常结果返回 ok 为 true 的对象；catch 首先调用 `signal.throwIfAborted()`，如果没有取消，则返回 ok 为 false 的结构。这样单条业务错误变成“正常完成的一项结果”，任务池不会触发批次取消。最后断言数组长度不变、每条原始下标仍能对应结果，并用总超时验证主动取消没有被包装层吞掉。下面给出实现上述合同的完整参考文件。 ```js settled-limit.mjs // settled-limit.mjs import assert from 'node:assert/strict'; import { setTimeout as delay } from 'node:timers/promises'; async function mapSettledLimit(items,limit,mapper,{signal}={}) { if (!Array.isArray(items)) throw new TypeError('items 必须是数组'); if (!Number.isInteger(limit) || limit<1) throw new RangeError('limit 必须为正整数'); signal?.throwIfAborted(); const output=new Array(items.length);let cursor=0; async function worker() { while(true) { signal?.throwIfAborted(); const index=cursor++; if(index>=items.length) return; try { const value=await mapper(items[index],index,signal); signal?.throwIfAborted(); output[index]={ok:true,value}; } catch(error) { // 取消不应伪装成某一条普通业务失败。 signal?.throwIfAborted(); output[index]={ok:false,error:{code:'ITEM_FAILED',message:'该片段处理失败'}}; } } } const workers=Array.from({length:Math.min(limit,items.length)},()=>worker()); const states=await Promise.allSettled(workers); const failed=states.find(state=>state.status==='rejected'); if(failed) throw failed.reason; signal?.throwIfAborted(); return output; } let active=0;let peak=0; async function preview(value,index,signal) { active+=1;peak=Math.max(peak,active); try { await delay(20,undefined,{signal}); if(value==='bad') throw new Error('模拟解析错误'); return `${index}:${value.toUpperCase()}`; } finally {active-=1;} } const result=await mapSettledLimit(['first','bad','last'],2,preview); assert.deepEqual(result.map(x=>x.ok),[true,false,true]); assert.equal(result[2].value,'2:LAST'); assert.equal(peak,2);assert.equal(active,0); const controller=new AbortController(); const timer=setTimeout(()=>controller.abort(new Error('用户取消')),5); try { await assert.rejects(()=>mapSettledLimit(['a','b','c'],2,preview,{signal:controller.signal}),/用户取消/); } finally {clearTimeout(timer);} assert.equal(active,0); assert.deepEqual(await mapSettledLimit([],2,preview),[]); console.log('逐项结果保留顺序；取消终止批次；活动任务归零'); ```
## 可验证的验收标准 把任务耗时打乱后结果顺序仍与输入一致；任何时刻 active 不超过 limit；失败和超时后 active 回到零；limit 为零时立即失败；空输入返回空数组；已经取消的 signal 即使面对空数组也应拒绝。能解释为什么本例实现了并发控制，却没有实现供应商每分钟额度限制。 ## 三个自测问题与答案 1. Promise.all 拒绝后其他请求自动停止吗？答案：不会，必须显式传递并响应取消信号。 2. 为什么不先创建三百个 fetch Promise 再分组 await？答案：请求通常已经启动，分组等待没有限制启动数量。 3. 用户看到超时是否证明外部操作失败？答案：不证明，外部系统可能已经完成，应通过幂等键、任务号或结果查询消除不确定性。 ## 本章验证记录 编写时使用 Node 22.22.0 对本章全部 3 个 JavaScript 完整文件执行了语法检查。已在本机实际运行通过：`concurrency.mjs`、`race-timeout.mjs`、`settled-limit.mjs`。 ## 本章官方参考 - [Node AbortController 与 AbortSignal](https://nodejs.org/docs/latest-v24.x/api/globals.html)：取消、超时与信号组合。 - [Node Timers](https://nodejs.org/api/timers.html)：Promise 定时器与取消错误。 - [ECMAScript Promise.all](https://tc39.es/ecma262/multipage/control-abstraction-objects.html#sec-promise.all)：聚合 Promise 的规范语义。