异步、并发与取消

本章解决什么问题#

一个 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 定义了这些方法。node:timers/promises 的定时器支持 signal，取消时通常拒绝为 AbortError；不同 API 的错误形状并非完全一致，不能只凭一个字符串推断全部失败原因。

我们设计的 mapLimit(items, limit, mapper, { signal }) 返回与 items 等长、同顺序的结果数组。limit 必须为正整数；mapper 收到当前项、原下标与组合信号；一项失败会取消同批任务，并在收尾后拒绝。这个合同没有实现每秒请求数限制：并发二且每项只需一毫秒，仍可能产生很高的请求频率。

完整示例：观察并发上限与超时#

环境：Node 22.22 或 24；无依赖。保存为 concurrency.mjs，执行 node 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，无依赖。实验只修改本地计数器，不发网络请求；它模拟客户端已经收到超时，但后台副作用仍然发生。

// 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 已取消则重新抛出；错误输出保留稳定代码，不向用户返回内部堆栈。

// 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：取消、超时与信号组合。
• Node Timers：Promise 定时器与取消错误。
• ECMAScript Promise.all：聚合 Promise 的规范语义。