异步控制流与并发管理

Promise 让异步结果可以组合，await 让依赖关系可以按同步代码的样子表达。但它们只回答“怎样等待结果”，从来没有回答“应该同时启动多少任务”。当一批 Promise 被一次性创建时，语法完全正确，数据库、内存和下游容量却可能先被耗尽。

于是这个 Node.js 服务遇到了第二次批量任务事故。上一次是任务没有被等待，这一次则是所有任务都被启动得太快。

我第一次意识到“异步写对了”和“系统扛得住”完全是两回事，是在一个批量导入接口上。

接口收到一批数据后，用 Promise.all(rows.map(saveOne)) 并发写库。测试环境里一次只有几十条，速度很漂亮；真实请求一下塞进上万条后，接口先是变慢，随后开始超时，数据库连接池等待数持续升高，进程内存也一路往上爬。最迷惑的是：客户端早已收到超时，数据库还在继续忙，日志又刷了好几分钟。

我最初怀疑过慢 SQL、数据库抖动，甚至怀疑垃圾回收暂停。逐项排查后才发现，真正的起点就是那行看似利落的 Promise.all：它在很短时间内启动了全部任务。连接池只能让少量查询真正执行，剩下的 Promise、参数和超时计时器全堆在进程与驱动队列里。客户端重试后，同一批工作又叠了一层。

把并发数降下来，错误率立刻回落，但事故还没有彻底结束：等待队列仍可能无限增长，超时也不等于取消，重试还可能重复写入。这个坑最终逼着我把异步控制流拆成几个必须明确回答的问题：任务什么时候启动、同一时刻跑多少个、生产者太快怎么办、结果没人要时怎样停下、重复执行会不会改坏数据。

第一处误判：把“异步”理解成“越多越快”

复盘时，我们先把几个经常混用的词摊开。它们不是术语洁癖，而是决定代码会怎样启动任务。

• 串行：任务乙要等任务甲完成后才启动。它描述的是启动顺序。
• 并发：一段时间内有多个任务都处于“尚未完成”的状态。它们的等待过程可以重叠。
• 并行：同一时刻确实有多个执行单元在做计算或 I/O。操作系统、数据库、远端服务以及 libuv 工作线程池都可能提供并行能力。
• 并发上限：允许同时处于运行状态的任务数量，例如最多保留 5 个在途请求。

假设三个独立请求各等待 1 秒。串行启动，总耗时接近 3 秒；一起启动，等待可以重叠，总耗时接近最慢的那一个。但任务从 3 个变成 1000 个时，“一起启动”就不再只是性能选择，而是在同时申请 1000 份资源。更稳妥的做法通常是只保留少量在途任务：先启动 5 个，任何一个完成后再补进下一个。

关键转折就在这里：Promise 组合器不是资源调度器。Promise.all() 只表达“一起等待”，不会替我们评估数据库连接数、文件描述符、堆内存或第三方接口配额。能不能同时启动，取决于任务关系；应该同时启动多少，取决于资源预算。

先看依赖，再决定是否串行

前一步的输出是后一步的输入

“先把地址转换成坐标，再根据坐标查询天气”就是一条天然的串行链。天气查询在拿到经纬度之前没有合法输入，提前启动没有意义。

回调、Promise 链和 await 都能表达这段依赖。写法可以变化，真正不能省略的是“第二步必须等第一步产出输入”这个约束。

async function buildWeatherReport(address) {
  const location = await geocode(address);
  const forecast = await getForecast(location.latitude, location.longitude);
 
  return {
    location: location.name,
    forecast,
  };
}

每个 await 只暂停当前 async 函数，不会让整个 Node.js 进程停住。等待地理编码时，事件循环仍然可以处理其他连接和回调。这里的“串行”只约束这条业务链，并不等于整个服务一次只能做一件事。

对一组数据逐个处理

数组长度在运行时才知道时，for...of 配合 await 是最直白的串行写法。下面的任务会严格按输入顺序开始；某个任务失败后，异常向外抛出，后面的任务不再启动。

import { setTimeout as delay } from 'node:timers/promises';
 
async function saveOne(record) {
  await delay(80);
  console.log(`已保存：${record.id}`);
}
 
async function saveInOrder(records) {
  for (const record of records) {
    await

我现在只在理由足够明确时选择串行：

• 后一步依赖前一步的结果；
• 外部系统要求严格顺序，例如迁移脚本必须先建主记录再建从记录；
• 操作共享且不可并发使用的资源；
• 业务主动把吞吐降到 1，避免触发旧系统的限制。

一个伪装成串行的陷阱：forEach(async () => {})

forEach 不会等待回调返回的 Promise。它会同步调用每个回调，于是所有异步任务很快都被启动；外层函数也不会因为这些回调而自动等待。

// 看上去像逐个保存，实际上会快速启动全部任务。
records.forEach(async (record) => {
  await saveOne(record);
});
 
console.log('这里可能先打印');

如果要串行，就用 for...of。如果确实要同时启动全部任务，就明确写成 Promise.all(records.map(saveOne))。这两种写法都不神秘，危险的是代码表现出的意图与真实启动方式不一致。

固定步骤与动态步骤

固定的两三步依赖链，用连续 await 最清楚；动态数组用循环更直观。reduce 也能把数组折叠成 Promise 链，但它常常只是把一段简单循环藏进更难读的表达式。

串行的代价是等待时间相加。它适合真实依赖，不适合被当成通用的“求稳开关”。三个互不相关的查询如果也逐个等待，只会平白拉长响应时间。

第二处转折：await 不是并发变慢的罪魁祸首

连续 await 和一起创建 Promise 的差别

下面两段代码最后都拿到三个结果，但时间线完全不同。

// 串行：前一个完成，后一个才开始。
const user = await fetchUser();
const orders = await fetchOrders();
const recommendations = await fetchRecommendations();

// 并发：三个函数先被调用，三个任务都已启动，再统一等待。
const userPromise = fetchUser();
const ordersPromise = fetchOrders();
const recommendationsPromise = fetchRecommendations();
 
const [user, orders, recommendations] = await Promise.all([
  userPromise,
  ordersPromise,
  recommendationsPromise,
]);

await 并不天然等于串行。真正决定时间线的是 Promise 在什么时候被创建。先等完一个再调用下一个，就是串行；先创建多个 Promise，再统一等待，就是并发。排查慢接口时，我会先找“任务启动点”，而不是看代码里出现了多少个 await。

四种组合器，其实是四种业务承诺

四个常用 Promise 组合器不是语法口味，它们分别回答“怎样才算这一批任务完成”。

Promise.all：必须全部成功

所有子任务都 fulfilled 时，结果 Promise 才 fulfilled，返回值按输入顺序排列，而不是按完成顺序排列。任何一个子任务 rejected，结果 Promise 立即 rejected。

const [profile, permissions, settings] = await Promise.all([
  loadProfile(userId),
  loadPermissions(userId),
  loadSettings(userId),
]);

它适合“缺一块就无法继续”的聚合结果、必须同时拿齐的配置，以及可以整体重试的无副作用读取。

这里的“立即 rejected”只描述外层 Promise 的状态。已经启动的其他任务通常还会继续运行，Promise.all 不会自动撤销网络请求或数据库查询。事故里“接口已经报错，数据库仍然忙”正是这个差异造成的。

Promise.allSettled：每一项都要有交代

它会等全部任务落定，再返回与输入一一对应的结果对象。成功项有 value，失败项有 reason。

const settled = await Promise.allSettled(
  userIds.map((id) => sendNotification(id))
);
 
const succeeded = settled.filter((item) => item.status === 'fulfilled');
const failed = settled.filter((item) => item.status === 'rejected');

它适合批量通知、批量校验、离线导入等允许部分成功的任务。拿到数组不是终点：还要记录失败原因、决定哪些错误可重试，并为成功项保存进度。

Promise.any：只要一个成功答案

它忽略先到的失败，返回第一个 fulfilled 的结果。只有所有任务都失败时，它才以 AggregateError rejected。

const fastestAvailable = await Promise.any([
  fetchFromReplica('https://replica-a.example/data'),
  fetchFromReplica('https://replica-b.example/data'),
  fetchFromReplica('https://replica-c.example/data'),
]);

它适合从多个等价副本中“谁先成功用谁”。拿到答案后仍要取消其余请求，否则连接和流量还在继续消耗。

Promise.race：谁先落定听谁的

第一个 settled 的任务决定结果；它可能成功，也可能失败。race 常被用来实现表面超时，但它只决定调用方先看到什么结果，不会自动停止输掉比赛的任务。

在写组合器之前，先把失败语义说清楚

“并发执行这一批”不是完整需求。我后来会先追问：

• 一个失败后，尚未开始的任务还要不要启动？
• 已经在运行的任务要不要取消？
• 已成功的结果是否保留，还是整体作废？
• 失败项是立即重试、延迟重试，还是进入死信队列？
• 调用方需要原始顺序、完成顺序，还是只要统计数据？

常见批处理大致可以归为三类：

如果任务有扣款、发券、发送消息之类的外部副作用，简单地“整体重试”会再次执行已经成功的项目。此时失败策略必须和幂等设计一起考虑，Promise 的状态无法替业务保证只执行一次。

事故止血：只在有空位时启动任务

Promise.all(items.map(task)) 的隐藏成本

map 会先遍历完整数组并调用每个 task，然后 Promise.all 才开始等待。假设数组里有 5 万个 URL，这段代码会在很短时间内创建 5 万个 Promise，并尝试发出大量请求。

// 数据量很大时，不要默认这样写。
const results = await Promise.all(urls.map(fetchJson));

压力会同时落在套接字和文件描述符、DNS 与 TLS 握手、数据库连接池、libuv 工作线程池、堆内存、第三方配额，以及等待处理回调的事件循环上。即使下游最终拒绝请求，本机也已经为“发起请求并等待失败”付出了成本。

并发上限把“任务总数”和“在途任务数”分开。总共有 5 万项，不代表必须同时运行 5 万项。上限为 8 时，真正调用异步函数的最多只有 8 个，其余项只是等待被领取的数据。

固定批次为什么仍会浪费空位

最容易想到的限制方法是每 10 项切成一批，对每批执行 Promise.all。它能限制并发，却存在批次屏障：如果这一批有 9 个任务很快、1 个任务很慢，9 个空位会一直等到最慢项完成，下一批才启动。

滑动任务池会更充分地利用空位：只要任意任务完成，马上从队列补进一个。可以把它想成几个长期工作的消费者，共同领取下一项。

把限制落在真正的启动点

下面的 mapWithConcurrency 是我更愿意放进实际项目的最小版本，它有几个关键性质：

• 只有 worker 真正取到一项时才调用 mapper，不会预先启动全部任务；
• 结果位置和输入位置一致；
• 支持“失败即停”和“收集全部”两种模式；
• 失败即停时，不再领取新任务，并通过 AbortSignal 尝试取消在途任务；
• 支持调用方主动取消。

代码可以保存为 pool.mjs，用现代 Node.js 直接运行。

import { setTimeout as delay } from 'node:timers/promises';
 
function createAbortError(message = '任务已取消') {
  const error = new Error(message);
  error.name = 'AbortError';
  return error;
}
 
function forwardAbort(source, targetController) {
  if (!source) return () 

这个实现把并发上限落在 mapper 的调用点上，因此不会像 items.map(mapper) 那样提前启动全部任务。多个 worker 共用 cursor；“读取索引并加一”是同一个同步片段，在执行到下一个 await 之前不会被另一个 JavaScript 回调插入，所以不会领取到同一项。

失败即停不等于瞬间停下

当一个任务失败时，池子可以马上停止领取新任务；已经开始的任务能否停下，则取决于 mapper 里的 API 是否接受并正确处理 AbortSignal。不支持取消的库仍会继续工作到自然结束。

这也是为什么函数不能只接收 () => Promise：把 { signal } 传给每个任务，取消能力才能沿调用链传到 fetch、定时器、流或你自己的异步函数。

并发上限怎样选

没有适用于所有系统的“最佳并发数”。我通常先找最紧的资源上限，再给在线流量和抖动留出余量：

• 数据库连接池有 20 个连接，而同一进程还要服务在线请求，离线批任务不能独占 20 个；
• 第三方接口允许每秒 50 次，不代表可以瞬间打出 50 个并持续保持 50 个在途请求；
• 单个任务会占用 30 MB 中间数据时，并发 100 在内存上就很危险；
• 任务主要占用 libuv 工作线程池时，提高到很大数值只会增长排队，不会线性提速。

先从偏小的值开始，观察吞吐、尾延迟、错误率、内存和连接池等待时间，再逐步调整。并发不是越高越快；一旦越过系统拐点，排队、超时和重试会让吞吐下降、延迟反而上升。

并发降到 8，为什么内存还在涨

并发限制仍可能留下无限队列

止血后还有一个现象让我困惑：数据库恢复了，进程内存却仍在缓慢上涨。原因是任务池最多运行 8 项，只保护了运行区；上游仍在不断读取数据并塞进内存队列。如果生产者每秒加入 1 万项，消费者每秒只能处理 100 项，等待队列仍会以每秒 9900 项的速度增长。

背压处理的正是生产速度与消费速度不匹配：消费者没有容量时，生产者必须暂停、等待、降速或拒绝新数据，不能把“以后再处理”理解成“无限放进内存”。

可以把它理解成水管和水箱。水箱容量有限，出水长期慢于进水，阀门就必须关小；只扩大水箱只能延迟溢出，不能改变长期速率差。

Node.js 流已经把背压做进协议

对可写流调用 write(chunk) 时，返回 false 表示内部缓冲区已达到阈值。调用方应该暂停写入，等到 drain 事件再继续。

import { once } from 'node:events';
import { createWriteStream } from 'node:fs';
 
async function writeLinesWithBackpressure(lines, filePath) {
  const output = createWriteStream(filePath, { encoding: 'utf8' });
 
  try {
    for await (const line of lines) {
      const canContinue = output.write(

真实的流转换更适合使用 stream.pipeline() 或它的 Promise 版本。pipeline 会连接上游和下游，并统一处理背压、错误与销毁。手写 data 事件后不停 write，却忽略 false 返回值，是非常典型的内存增长成因。

普通任务队列也要设计背压

不是所有任务都以 Stream 表示。HTTP 接口收到批量作业后，可以根据业务价值选择：

• 队列未满时接收，达到容量后返回“系统繁忙”，让调用方稍后重试；
• 接收请求后只写入有容量控制的持久化队列，由消费者按能力领取；
• 上游和下游在同一个进程时，让 enqueue() 返回 Promise，直到队列腾出位置才 fulfilled；
• 对低价值任务采用丢弃、合并或只保留最新一项的策略。

关键不是一定要等，而是队列满时必须有明确动作。无限等待、无限缓存和无限重试，只是在把压力换成内存耗尽或级联故障。

并发、速率限制和背压不是一回事

一个接口完全可能同时需要三者：最多 8 个在途请求、每秒最多启动 20 个、等待队列最多 200 项。只限制并发，快速任务仍可能在一秒内启动很多次；只限制速率，慢请求又可能积累大量在途连接。三个数字保护的是三种不同的失控路径。

接口已经超时，为什么数据库还在忙

Promise.race 只决定谁先返回

下面的写法能让调用方在 2 秒后收到错误，看上去实现了超时，但 fetch 仍可能继续下载响应。

function timeoutAfter(ms) {
  return new Promise((_, reject) => {
    setTimeout(() => reject(new Error('请求超时')), ms);
  });
}
 
await Promise.race([
  fetch('https://example.com/slow'),
  timeoutAfter(2000),
]);

如果业务请求先完成，定时器还在；如果定时器先完成，网络请求也还在。事故现场那些“响应结束后仍在运行”的任务，就是这样变成孤儿工作的。它们继续占用连接、带宽、CPU 和日志空间。

真正有效的超时，需要把“结果已经没人要了”表达成取消信号，并把信号一路传给实际工作的 API。

用 AbortSignal 贯穿调用链

现代 Node.js 的 fetch 支持 AbortSignal。AbortSignal.timeout(ms) 可以创建超时信号；AbortSignal.any() 可以把“请求方主动取消”和“内部超时”组合起来。

async function fetchJson(url, { signal, timeoutMs = 2000 } = {}) {
  const timeoutSignal = AbortSignal.timeout(timeoutMs);
  const combinedSignal = signal
    ? AbortSignal.any([signal, timeoutSignal])
    : timeoutSignal;
 
  const response = await fetch(url, { signal: combinedSignal });
 
  if (!response.ok) {

取消是一套协作协议，不是强制终止任意 Promise 的开关。自己的异步函数需要主动检查 signal.aborted，或调用 signal.throwIfAborted()；注册 abort 监听器时使用 { once: true }，任务结束后移除不再需要的监听器。否则取消机制本身也会积累资源。

超时要分层

一个 HTTP 请求常常经历排队、DNS、建连、TLS、等待响应头、读取响应体和业务处理。只设一个总超时虽然简单，却很难判断时间究竟耗在哪里。

我更倾向于按预算分层设置：

• 单次下游调用的超时；
• 整个入口请求的截止时间；
• 任务在队列中的最长等待时间；
• 优雅停机时允许在途任务完成的宽限时间。

截止时间还要向下游传播。入口只剩 300 毫秒时，再启动一个通常需要 2 秒的调用没有价值。越靠近底层的组件越早知道“结果已经没人要了”，系统越能及时释放资源。

清理放在 finally

无论成功、失败还是取消，连接、临时文件、锁和计时器都要释放。finally 是统一收尾的位置。

async function withLease(pool, work) {
  const lease = await pool.acquire();
 
  try {
    return await work(lease);
  } finally {
    lease.release();
  }
}

不要只在成功路径释放资源，也不要把“调用方已经收到超时”误认为“底层资源已经释放”。

重试把请求救回来，也可能把数据写坏

await 会把一次操作分成多个时间片

止住容量问题后，还要面对更隐蔽的正确性问题。看下面的库存扣减：先读库存，经过一次异步等待，再写回新值。两个请求可能都读到 10，然后分别写回 9。系统处理了两次扣减，库存却只减少 1，这就是丢失更新。

async function unsafeDecreaseStock(productId) {
  const product = await db.products.findById(productId);
 
  if (product.stock <= 0) {
    throw new Error('库存不足');
  }
 
  product.stock -= 1;
  await db.products.save(product);
}

Node.js 的 JavaScript 回调通常单线程执行，并不代表业务没有竞态。只要操作跨越 await，其他请求就能在当前函数恢复前读取或修改同一份外部状态；多进程、多容器和其他数据库客户端还会扩大这个窗口。

原子条件更新守住状态

把“检查库存大于 0”和“库存减 1”交给数据库一次完成，避免读写之间留下窗口。下面是通用 SQL 思路：

UPDATE products
SET stock = stock - 1
WHERE id = $1 AND stock > 0
RETURNING id, stock;

如果没有返回行，就表示库存不足或商品不存在。多条记录的关联修改需要事务、合适的隔离级别，或根据业务模型使用乐观锁和版本号。进程内的普通变量锁保护不了其他进程。

幂等键处理重复请求

超时和重试还会制造另一类竞态：客户端发起扣款，服务端其实已经成功，但响应在网络中丢失；客户端于是重试同一个请求。如果服务端把重试当成新业务，用户就会被扣两次。

幂等的目标是：同一个业务意图执行多次，最终效果和执行一次相同。常见做法是让客户端为一次业务操作生成幂等键，服务端用数据库唯一约束登记它。

开始事务
  尝试插入（幂等键，处理中），幂等键有唯一索引
  如果键已存在：读取并返回之前的状态或结果
  如果插入成功：执行一次业务写入
  保存结果，把状态改为已完成
提交事务

幂等记录不能只放在进程内 Map 里。进程重启会丢数据，多实例之间也不共享。可靠的方案通常需要持久化存储、唯一索引、结果保留期限，以及“记录仍是处理中、执行者却已崩溃”时的恢复规则。

并发池不会自动消灭竞态

把并发上限设为 1，只能让这个池子里的任务串行。另一个进程、另一条队列或直接到达的 HTTP 请求仍可能同时修改数据。并发控制保护容量；原子操作、事务、唯一约束和幂等键保护正确性。不要拿容量开关代替数据约束。

重试之前先分类

适合重试的通常是短暂错误，例如连接被重置、下游限流或服务暂时不可用。参数错误、权限错误和业务校验失败不会因为多试几次就变好。

即使是短暂错误，也要同时满足：

• 操作本身幂等，或有幂等键保护；
• 重试有次数上限；
• 使用退避和随机抖动，避免大量客户端同时重试；
• 整体截止时间尚未耗尽；
• 最终失败有可追踪的记录。

缺少这些约束的自动重试，会把一次故障放大成多倍流量；如果操作不幂等，还会把容量事故升级成数据事故。

最后的线索藏在“还没完成”的任务里

异步 API 很容易制造一种错觉：函数立即返回 Promise，好像工作很轻。实际上，每个未完成任务背后都可能保留套接字、响应缓冲区、数据库连接、文件句柄、闭包变量、定时器和监听器。事故里真正拖垮进程的，从来不只是 Promise 对象，而是它代表的整套未完成工作。

资源通常从哪里被耗尽

文件描述符和套接字

一次启动过多文件或网络操作，可能触及进程或系统的文件描述符上限。HTTP 客户端的连接池、keep-alive 配置和单个源站的连接数，都要和并发策略一起考虑。

数据库连接池

业务并发大于连接池容量时，任务会在驱动内部排队。应用层再叠加一个无限队列，真实等待时间就被藏起来了。查询超时后如果数据库操作没有取消，连接仍可能长期被占用。

堆内存

大量 Promise、闭包捕获的数据，以及等待统一汇总的结果都会占用堆。Promise.all 会保留已经完成的结果，直到最慢任务结束并产出整个数组。处理大文件或海量记录时，边生产边消费通常比一次性聚合更稳。

事件循环和工作线程池

网络 I/O 主要由操作系统异步处理；部分文件系统、DNS、压缩和密码学操作会使用 libuv 工作线程池。把成千上万项提交给一个较小的线程池，只会形成更长的内部队列。同步 CPU 计算则会直接阻塞事件循环，让其他请求的回调无法及时执行。

复盘时，先找排队指标

这次排障让我形成了一个习惯：不仅看执行耗时，还要把排队本身做成指标。至少记录：

• 当前运行数、等待队列长度和队列等待时间；
• 每类下游请求的成功率、超时率和尾延迟；
• 数据库连接池的占用数与等待数；
• 取消次数、取消后仍继续运行的任务数；
• 重试次数与最终失败数；
• 进程堆内存、事件循环延迟和打开的句柄数量。

只看平均耗时很容易错过问题。队列可能已经越来越长，但少数刚完成的快任务仍让平均值显得正常。容量问题往往先出现在高分位延迟、连接池等待和队列长度上。

一套可复用的决策顺序

后来再遇到批量异步任务，我会按同一套顺序做设计。它比先挑一个 Promise API 更可靠。

如果只记一句话，可以记成：依赖决定顺序，资源决定并发，消费者能力决定背压，业务语义决定失败策略，数据约束决定并发正确性。

用三个问题检查是否真的掌握

把那次事故压缩成一张检查清单

回头看，最初那行 Promise.all(rows.map(saveOne)) 并没有语法错误，它只是隐含了几个危险假设：任务数量很小、下游容量无限、失败后其他任务继续运行也没关系、调用方超时后结果仍然有价值、重试不会产生副作用。

真正可靠的异步系统，需要主动拆掉这些假设：有依赖就串行，无依赖也要按资源预算并发；运行区用并发上限保护，等待区用背压保护；超时和失败通过 AbortSignal 尽量停止无用工作；共享状态交给原子更新、事务和唯一约束；有副作用的操作用幂等键守住重试；最后用队列长度、等待时间、尾延迟、取消和重试指标验证设计是否有效。

并发管理的目标从来不是“同时跑得越多越好”，而是让系统在压力变大、下游变慢、请求被取消时，仍然知道该启动多少工作、该拒绝什么、该停下什么，以及怎样保证数据没有被重复或错乱地修改。

这些规则在批处理任务里很直观：数组是生产者，任务池是运行区，数据库是下游。放到 Web 服务里，生产者会变成不断到来的 HTTP 请求和请求体，结果还要沿着一条可能很慢、也可能提前断开的网络连接返回。并发、背压、超时和取消接下来都会进入同一个更具体的生命周期：从收到请求头，到响应真正写完。