实现一个限制最大并发数的任务调度器。

核心回答

可以用固定数量的异步 worker 消费任务队列：最多启动 limit 个 worker，每个 worker 一次只执行一个任务，等它成功或失败后立即领取下一项。这样不会等一整批任务都结束才补任务，也不会同时把所有请求发出去。

再用一个递增下标分配任务，并按原下标保存结果。下面约定：输入是稳定的有限数组，单项失败继续执行其他项，全部结束后返回每项的成功值或失败原因，顺序与输入一致。并发数必须是正整数；空任务列表返回空数组。

JavaScript 实现

worker(item, index) 在领到任务后才被调用，可以返回普通值或 Promise。这里控制的是每项 worker 调用所代表的完整任务，不是 worker 内部自行启动的其他后台操作。

async function asyncPool(limit, items, worker) {
  if (!Number.isInteger(limit) || limit < 1) {
    throw new RangeError("limit 必须是正整数");
  }
  if (!Array.isArray(items) || typeof worker !== "function") {
    throw new TypeError("需要任务数组和 worker 函数");
  }

  const total = items.length;
  const results = new Array(total);
  let next = 0;

  async function run() {
    while (next < total) {
      const index = next++;
      try {
        const value = await worker(items[index], index);
        results[index] = { status: "fulfilled", value };
      } catch (reason) {
        results[index] = { status: "rejected", reason };
      }
    }
  }

  const workers = Array.from({ length: Math.min(limit, total) }, () => run());
  await Promise.all(workers);
  return results;
}

每个 worker 在 await 前领取唯一的下标，完成后才回到循环领取下一项。在同一 JavaScript 执行环境中，下标的同步读取与递增不会被另一个 worker 插入执行；这里不涉及多个操作系统线程共享这个计数器。

try/catch 同时处理 worker 同步抛错和返回 Promise 后的拒绝。失败被记录为结果，因此不会让某个 worker 提前退出，后面的任务仍能补进来。外层 Promise.all 等待的是有限个 worker 的完整消费过程。

用于批量请求

请求地址作为普通数据传入，真正的请求在 worker 获得执行机会时创建：

const results = await asyncPool(3, urls, async (url) => {
  const response = await fetch(url);
  if (!response.ok) {
    await response.body?.cancel();
    throw new Error(`HTTP ${response.status}`);
  }
  return response.json();
});

fetch 收到 404、500 等响应时不会自动拒绝，所以这里显式检查 response.ok，错误时关闭未读取的响应体。任务还包括响应体读取和 JSON 解析，发生解析错误也会记录为该项失败；并发名额不会在刚收到响应头时就提前释放。

如果传入的是一组任务函数，也可以让 worker 调用对应函数。关键是延迟启动：先把所有 fetch() 调好，再把产生的 Promise 交给调度器，已经无法限制那些请求的启动数量。

为什么既受限，又能及时补位？

最多只有 min(limit, total) 个 worker，每个在当前任务结束前都不会领取下一项，因此受调度器管理的在途任务不超过这个数量。

例如并发上限为 2，先执行下标 0、1。若 1 先结束，它所在的 worker 立即执行 2，不必等待 0。结果仍写到 results[1]，而不是 push 到数组末尾，所以返回顺序不受完成先后影响。

调度本身对每项进行一次领取和一次结果写入，额外组织开销为 O(n)，结果与 worker 状态需要 O(n + limit) 空间；实际总耗时还取决于任务耗时及依赖的服务，不能简单保证等于串行耗时除以并发数。

常见追问

为什么不用 Promise.all 或 forEach(async ...)？

Promise.all(items.map(worker)) 会先对全部输入调用 worker，再等待结果，本身没有最大并发控制。forEach(async ...) 同样不会等待前一个回调完成，也不会汇总等待整个过程。需要由队列、worker 数量或显式在途集合来限制何时启动。

失败时想立即返回，怎么调整？

先明确是“总体 Promise 立即失败”，还是“停止领取新任务”，以及是否需要取消在途任务。三者是不同动作。立即 reject 不会自动取消已经发出的请求；停止标记只能阻止后续领取，真正取消请求还要把 AbortSignal 等机制传给执行函数。

给任务加 Promise.race 超时，就一定保持真实请求并发上限吗？

不一定。超时 Promise 先结束，并不代表底层请求已经停止。如果立刻释放名额再启动新请求，真实在途请求可能越积越多。应尽可能取消底层操作并确认结束，或继续把尚未结束的操作计入资源预算。

某个任务永远不结束怎么办？

它会一直占用对应 worker，整体结果也无法全部完成。实际系统需要为相应 I/O 配置超时、取消或其他结束机制，并明确超时算失败还是允许重试。重试也应占用受控名额，避免绕过并发限制。

参考资料