人人都会AI编程

4.4 异步并发控制：串行、并行、限流、竞态处理

更新时间：2026-07-10

在 4.3 节中我们看到 async/await 让异步代码拥有了同步式的书写体验，但这仅仅是基操。实际开发中，我们经常需要同时处理一堆异步任务，比如批量发送请求、并发读取文件、轮询多个数据源等。这时候真正考验功力的，是如何控制这些异步任务的执行顺序和并发数量。本节我们将拆解四种最常见的并发模式：串行、并行、限流和竞态，每种都配有实用的代码模式和踩坑经验。

4.4.1 串行执行：一个接一个，稳中有序

串行是指多个异步任务按照严格的先后顺序执行：前一个完成之后才开始下一个。这通常适用于任务之间存在依赖关系，或者需要保证执行顺序的情景。

使用 async/await 实现串行

最简单的串行就是在一个 async 函数中依次 await：

async function serialTasks(tasks) {
  const results = [];
  for (const task of tasks) {
    // 等待每个任务完成再继续
    results.push(await task());
  }
  return results;
}

// 示例：按顺序查询数据库
async function getUserOrders(userId) {
  const user = await db.query('SELECT * FROM users WHERE id = ?', [userId]);
  const orders = await db.query('SELECT * FROM orders WHERE user_id = ?', [user.id]);
  return { user, orders };
}

使用数组 reduce 实现串行链

如果任务数量是动态的，可以用 reduce 构建一个顺序执行的 Promise 链：

function runSerial(tasks) {
  return tasks.reduce((prevPromise, task) => {
    return prevPromise.then(results => {
      return task().then(result => [...results, result]);
    });
  }, Promise.resolve([]));
}

// 使用
runSerial([task1, task2, task3]).then(results => {
  console.log('所有任务顺序完成', results);
});

当然，这种写法没有 async/await 直观，而且错误处理需要额外小心，建议优先使用 for...of + await。

什么时候必须串行？

  • 后一个任务依赖于前一个任务的输出（如先获取用户ID，再查订单）。
  • 操作共享资源，需要互斥访问（如写入同一个文件，或者更新同一行数据库记录）。
  • 需要遵守外部 API 的调用顺序限制（如某些支付接口要求业务流水号递增）。

如果任务之间没有依赖，串行会白白浪费等待时间，应该果断改用并行。

4.4.2 并行执行：同时发起，最快聚拢

并行是指一次性启动所有异步任务，让它们各自独立运行，最后统一收集结果。这种模式能最大化利用 I/O 等待时间的重叠，是提升吞吐量的主要手段。

Promise.all – 一荣俱荣，一损俱损

const [user, articles, settings] = await Promise.all([
  fetchUser(),
  fetchArticles(),
  fetchSettings()
]);

当所有任务都成功时，all 返回的结果数组顺序与输入的任务顺序一致。但如果任何一个任务 reject，整个 Promise.all 立即 reject，并丢弃其他任务的结果。这种“快速失败”特性适合需要数据完整性的场景。

Promise.allSettled – 求全责备，逐个检查

如果希望即便部分任务失败，也要拿到所有任务的结果（包括成功和失败），用 allSettled：

const results = await Promise.allSettled([
  fetchUser(),
  fetchArticles(),
  fetchSettings()
]);

results.forEach((result, index) => {
  if (result.status === 'fulfilled') {
    console.log(`任务${index}成功`, result.value);
  } else {
    console.error(`任务${index}失败`, result.reason);
  }
});

这在聚合多个独立数据源时非常有用，比如从三个不同的天气 API 获取数据，哪怕某一个挂了，我们还是可以用另外两个。

并行任务的数量限制

并行虽然强大，但并不是可以无脑堆叠的。如果一次启动数百个网络请求，可能会触发操作系统的文件描述符上限，或者被目标服务器限流甚至封 IP。此时就需要引入限流机制。

4.4.3 限流（并发控制）：管住并发的“阀门”

限流（Concurrency Limiting）指在保持异步任务并行的同时，将同时执行的任务数限制在一个安全范围内。就像银行柜台，就算有 100 个客户，也只能同时开放 5 个窗口服务，其他人排队等待。

我们通常不需要从零实现限流，p-limit 和 async-pool 等成熟库已经非常轻量且可靠。

使用 p-limit（推荐）

p-limit 是最简单的选择：

npm install p-limit
const pLimit = require('p-limit');
const limit = pLimit(5);  // 最多同时执行 5 个任务

const tasks = urls.map(url => {
  return limit(() => fetch(url));  // 包装每个异步操作
});

const results = await Promise.all(tasks);

p-limit 维护了一个内部队列，确保同一时间只有指定数量的 limit 回调在运行，其余排队等待。当某个任务完成，立即开启下一个。

手动实现一个简单的并发池

理解背后的原理有助于定制需求。基本的思路是：用一个数组保存待执行的任务，用一个计数器记录当前正在执行的数量，并通过 Promise 控制流程。

async function asyncPool(concurrency, tasks, iteratorFn) {
  const ret = [];                 // 存储所有结果
  const executing = new Set();   // 跟踪正在执行的 Promise

  for (const item of tasks) {
    const p = Promise.resolve().then(() => iteratorFn(item));
    ret.push(p);

    if (concurrency <= tasks.length) {
      // 当正在执行的数量达到上限，就从 executing 中移除最先完成的那个
      const e = p.then(() => executing.delete(e));
      executing.add(e);
      if (executing.size >= concurrency) {
        await Promise.race(executing);
      }
    }
  }

  return Promise.all(ret);
}

// 用法：每次最多并发 3 个上传任务
const results = await asyncPool(3, files, file => upload(file));

这段代码的核心是用 Promise.race(executing) 等待任何一个任务完成，从而腾出位置。在实际项目中更推荐用社区库，因为它们的边界处理更加完善。

限流的典型适用场景

  • 批量调用第三方 API，对方有 QPS 限制（例如每分钟最多 100 次）。
  • 并发写入同一个数据库，需要控制连接池大小。
  • 大文件分块上传，需要避免占用过多网络带宽。
  • 避免打开过多文件描述符导致 EMFILE 错误。

4.4.4 竞态处理：唯快不破，但要善后

竞态（Race）指的是启动多个异步操作，只取第一个完成的结果，其余的全部抛弃。最常见的用途是实现超时机制，或者从多个数据源中获取最快的响应。

Promise.race – 只认最快的那个

// 为 fetch 添加 3 秒超时
function fetchWithTimeout(url, timeout = 3000) {
  return Promise.race([
    fetch(url),
    new Promise((_, reject) => 
      setTimeout(() => reject(new Error('请求超时')), timeout)
    )
  ]);
}

任何一方先 settled（无论成功还是失败），race 就立刻结束，另一方即使还在执行中也不会再被处理。但要注意，未被采用的那个异步操作并不会被取消，它仍然会在后台继续执行并消耗资源。在 Node.js 中如果使用了网络请求或定时器，应该有额外的清理逻辑。

使用 AbortController 实现真正的取消

在较新的 Node.js 版本中，fetch 支持 AbortController：

const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), 3000);

try {
  const response = await fetch(url, { signal: controller.signal });
  clearTimeout(timeoutId);
  // 处理响应
} catch (err) {
  if (err.name === 'AbortError') {
    console.log('请求被主动取消');
  }
}

对于非 fetch 的异步操作（如数据库查询），则需要各自对应的取消机制。

Promise.any – 取第一个成功的，忽略失败

Promise.any 是 ES2021 引入的，与 race 的不同在于：它会忽略 reject，直到有一个成功或全部失败。这特别适合多源容灾：

const result = await Promise.any([
  fetchFromCDN(),
  fetchFromBackupCDN(),
  fetchFromOrigin()
]);

只要任何一个源成功，我们就立即使用它的结果。如果全部失败，会抛出一个 AggregateError。

竞态场景的注意事项

  • 资源清理：未选中的任务可能持有数据库连接、文件句柄，如果不再需要，应主动释放。
  • 不死任务：使用 setTimeout 模拟超时时，如果被保护的任务已经完成，但超时定时器还在，应该 clearTimeout，防止内存泄漏。
  • 竞态条件：避免多个并发操作对同一数据进行无顺序保证的读写，容易产生状态不一致。

4.4.5 综合运用：真实场景下的并发策略

实际业务往往是几种模式的混合体。举个例子：用户需要导出最近 30 天的订单报表，每天的数据需要单独查询且没有依赖，但数据库连接池只允许同时跑 10 个查询。这时候可以限流并行：

const limit = pLimit(10);
const dailyPromises = days.map(day => 
  limit(() => queryDailyOrders(day))
);
const dailyResults = await Promise.all(dailyPromises);
// 合并结果,生成报表

再比如，一个页面需要同时加载用户信息、评论列表和推荐内容，其中评论和推荐可以失败不影响整体展示，但用户信息是必须的。我们可以先用 Promise.all 获取用户信息，再用 Promise.allSettled 获取可降级的数据：

const user = await fetchUser();
const [comments, recommendations] = await Promise.allSettled([
  fetchComments(),
  fetchRecommendations()
]);
// user 有问题直接失败，comments 和 recommendations 失败则用空数组替代

掌握这四种异步并发控制模式，你就能从容应对大多数 I/O 密集型场景中的性能与可靠性挑战。在第七篇的“性能优化”和“微服务实战”章中，我们还会继续用这些模式来构建更高阶的服务架构。