人人都会AI编程

并发限流实现、批量请求控制

更新时间：2026-07-11

在现实的后端开发中，经常会遇到需要同时发起大量异步操作的情况：调用第三方 API、批量查询数据库、处理文件等。如果直接使用 Promise.all 一次性启动所有异步任务，可能会瞬间把下游服务或数据库的连接池打满，甚至触发对方的限流机制。更常见的需求是：控制并发的请求数量，一批一批地处理，既能保持速度，又不会压垮资源。

什么是并发限流

并发限流的核心思路是：设定一个最大并发数（比如同时只能有 5 个请求在执行），每当一个请求完成，就从待处理任务队列中取出下一个开始执行，保证同一时刻正在运行的任务数不超过设定的上限。

这种方式在以下场景中非常重要：

  • 调用第三方开放 API，对方限制了每秒的请求次数（Rate Limit）；
  • 数据库查询过多会导致连接池耗尽或锁竞争；
  • 大量文件上传/下载时需要保护网络和磁盘带宽；
  • 避免由于一下子创建太多 Promise 而导致内存占用过高。

手动实现并发控制器

Node.js 生态中已经有非常成熟的并发控制库，如 p-limit、async、bottleneck 等，但理解其底层实现有助于在特殊场景中自己编写。下面是一个经典的“并发池”实现：

/**
 * 并发执行异步任务，限制最大并发数
 * @param {Array} tasks 返回 Promise 的函数数组
 * @param {number} limit 最大并发数
 * @returns {Promise<Array>} 按原顺序返回每个任务的结果
 */
async function asyncPool(tasks, limit) {
  const results = [];
  const executing = new Set(); // 追踪当前正在执行的 Promise

  for (const task of tasks) {
    // 将每个任务包装成一个 Promise
    const promise = Promise.resolve().then(() => task());
    results.push(promise);
    executing.add(promise);

    // 执行完后从集合中移除
    const clean = () => executing.delete(promise);
    promise.then(clean, clean);

    // 如果达到并发上限，等待其中任意一个完成
    if (executing.size >= limit) {
      await Promise.race(executing);
    }
  }

  // 所有任务发起后，等待剩余的任务跑完
  return Promise.all(results);
}

使用示例：

const tasks = [
  () => fetchUser(1),
  () => fetchUser(2),
  () => fetchUser(3),
  () => fetchUser(4),
  () => fetchUser(5),
  () => fetchUser(6),
  // ... 更多任务
];

// 同时最多只保持 3 个请求在执行
const results = await asyncPool(tasks, 3);
console.log(results);

这里的关键是维护一个 executing Set，当并发数达到 limit 时，使用 Promise.race 等待任意一个完成，然后循环自然进入下一个任务。这个实现适用于任务总数已知、按顺序执行结果的场景，而且不限任务类型，只需要它们返回 Promise。

使用 p-limit 库实现并发限流

p-limit 是目前 Node.js 社区中用得最广泛的并发限制工具，原理与自定义的 asyncPool 类似，但封装得更加健壮，支持错误处理、自定义队列等。

npm install p-limit
const pLimit = require('p-limit');
const limit = pLimit(5); // 最大并发数 5

const tasks = userIds.map(userId => {
  return limit(() => fetchUser(userId));
});

const results = await Promise.all(tasks);

p-limit 会返回一个限流函数，每次调用 limit(fn) 都会等待有空位后再执行 fn，从而将并发数总控制在设定范围内。如果是动态生成任务（比如从一个无限列表读取），也可以像使用普通函数一样使用：

for (const userId of userIds) {
  limit(async () => {
    const user = await fetchUser(userId);
    await saveUser(user);
  });
}

这种写法不会一次性创建所有 Promise，而是边消费边生产，内存占用更低。

批量请求控制：分批次处理大量数据

并发限流关注的是同一时刻运行的并行任务数，而批量请求控制则更侧重于将总任务按固定大小切割成多个批次，一批完成后再启动下一批。这在处理海量数据时尤为常用，例如对数据库进行分批更新、同步百万量级的数据等。

最简单的批量处理可以使用循环加 Promise.all：

async function batchProcess(items, batchSize, processor) {
  const results = [];
  for (let i = 0; i < items.length; i += batchSize) {
    const batch = items.slice(i, i + batchSize);
    const batchResults = await Promise.all(
      batch.map(item => processor(item))
    );
    results.push(...batchResults);
  }
  return results;
}

// 每批处理 100 条，逐批调用外部 API
const allResults = await batchProcess(largeArray, 100, async (item) => {
  return externalApi(item);
});

如果需要既限制并发数，又按批次处理，可以将两种技术结合起来。例如使用 p-limit 配合分批循环，或者直接改造 asyncPool 支持大数据量流式处理：

async function processWithLimit(items, limit, processor) {
  const results = [];
  const executing = new Set();

  for (const item of items) {
    const promise = Promise.resolve().then(() => processor(item));
    results.push(promise);
    executing.add(promise);
    promise.finally(() => executing.delete(promise));

    if (executing.size >= limit) {
      await Promise.race(executing);
    }
  }
  return Promise.all(results);
}

这样就既控制了并发，又不一次性把所有任务都加入队列（通过分片或使用迭代器等方式），能够处理超大量数据而不会撑爆内存。

实际场景中的注意点

  1. 错误处理

无论自定义实现还是使用 p-limit，都需要决定错误策略：是某个任务失败后立即停止整个队列（如 Promise.all 行为），还是记录错误并继续？通常我们会单独捕获每个异步任务的错误，以免整个批次都失败。可对每个任务用 .catch() 包装，返回一个标记错误的结果对象。

  1. 超时控制与重试

并发限流经常与超时重试机制搭配。如果下游 API 不稳定，可以对单个请求加上超时（Promise.race 与延时），并在失败时重试几次，同时注意重试也不能无限制，避免无限堆积。

  1. 动态增加任务

如果你需要根据前一阶段的结果动态生成后续任务（如分页拉取直到没有数据），那么并发控制必须支持在运行过程中动态地 limit(fn) 或推入队列。p-limit 天然支持这种模式，而自定义实现则需要不断循环检查空闲位置。

  1. 与数据库连接池的协同

即使你限制了并发请求数，底层的数据库连接池大小也需要匹配。如果并发数高于连接池的最大连接数，新任务会因拿不到连接而排队，可能反而降低吞吐量。通常建议并发限制数 ≤ 连接池最大容量。

  1. 优雅关闭

在长期运行的服务中，如果进程收到 SIGTERM 信号，需要等待当前正在执行的限流任务完成后再退出，而不是直接暴力中断。

总结

并发限流和批量请求控制是 Node.js 异步编程中的“节奏控制”工具箱。通过限制同时进行的异步操作数量，我们可以既充分利用 Node.js 的异步并发能力，又避免对下游造成过大压力。p-limit 等库让这一模式变得极其简单，而理解其背后的机制可以帮助你在更复杂的场景中灵活调整策略，在性能与可靠性之间找到最佳平衡点。