编码母题 / P0 / 异步编程 / 01

有序返回的并发任务池

先读清题意与边界，独立尝试，再展开思路并对照可运行的参考代码。

重要程度必练难度中等预计完成35–45 分钟

难度按 LeetCode 简单／中等／困难三档估计；工程题并非 LeetCode 官方题目。用时指独立编码并检查主要边界的参考时间，因经验而异。

题目是什么

输入尚未启动的任务函数数组和并发上限。任务按输入顺序领取，运行数不超上限，最终结果仍按输入顺序排列；任一失败后停止启动新任务，已运行任务无须强制取消。

输入输出示例

并发上限为 2，三个任务完成顺序即使是 B、A、C，结果仍须是 [A,B,C]。

练习起手式与验证

复制代码后补全空方法，再运行下方三组验证；复杂题所需的辅助类型与输入准备一并放在代码块中。

/*
 * 题目：有序返回的并发任务池
 * 要求：输入尚未启动的任务函数数组和并发上限。任务按输入顺序领取，运行数不超上限，
 * 最终结果仍按输入顺序排列；任一失败后停止启动新任务，已运行任务无须强制取消。
 * 示例：并发上限为 2，三个任务完成顺序即使是 B、A、C，结果仍须是 [A,B,C]。
 */
async function asyncPool(tasks, concurrency) {}

const sleep = (milliseconds) =>
  new Promise((resolve) => setTimeout(resolve, milliseconds));

(async () => {
  console.log(await asyncPool([], 2)); // 预期：[]
  console.log(await asyncPool([() => 1, () => 2, () => 3], 2)); // 预期：[1, 2, 3]

  const completionOrder = [];
  const orderedResult = await asyncPool(
    [
      async () => {
        await sleep(30);
        completionOrder.push("A");
        return "A";
      },
      async () => {
        await sleep(10);
        completionOrder.push("B");
        return "B";
      },
      async () => {
        await sleep(20);
        completionOrder.push("C");
        return "C";
      },
    ],
    2,
  );

  console.log(completionOrder); // 预期：["B", "A", "C"]
})().catch(console.error);

解题思路

展开思路、不变量与常见边界

关键切入点：固定数量 Worker 共享递增索引，结果写回原索引；失败标记阻止后续领取。

四、核心模型与不变量使用固定 Worker 模型：创建 min(concurrency, tasks.length) 个 Worker，每个 Worker 完成当前任务后，再通过共享索引领取下一项。必须始终成立： - 活动任务数不超过 Worker 数，也就不超过 concurrency。 - 共享索引只递增，每个任务最多领取一次。 - 结果写入原始索引，因此完成顺序不会改变结果顺序。 - 观察到失败后设置 stopped，其他 Worker 不再领取新任务。五、复杂度调度器处理每个任务一次，时间复杂度 O(n)；结果数组占 O(n)，Worker 占 O(min(n, concurrency))。任务自身耗时不计入算法复杂度。六、常见错误 - 接收 Promise 数组，导致任务在调度前已经启动。 - 使用 results.push，错误地返回完成顺序。 - 使用 forEach(async () => {})，外层无法等待回调。 - reject 后继续领取任务，或误认为 reject 会取消活动任务。 - 用 splice(0, 1) 破坏输入并反复移动数组元素。七、高价值变体 1. All-settled：记录每项成功或失败，单项失败不停止调度。 2. AbortSignal：停止新调度，并把取消信号传给支持协作取消的任务。八、工程迁移同一模型可用于批量请求、分片上传、文件处理、批量 Embedding、 Agent Tool 并行调用和 Eval 批处理。

参考代码与验证

“关键机制”注释用于解释实现，下载链接提供未经页面注释修改的原始 JS 文件。

// 关键机制：固定数量 Worker 共享递增索引，结果写回原索引；失败标记阻止后续领取。
async function asyncPool(tasks, concurrency) {
  if (!Array.isArray(tasks)) {
    throw new TypeError("tasks must be an array");
  }

  if (!Number.isInteger(concurrency) || concurrency < 1) {
    throw new RangeError("concurrency must be an integer >= 1");
  }

  for (let index = 0; index < tasks.length; index += 1) {
    if (typeof tasks[index] !== "function") {
      throw new TypeError(`tasks[${index}] must be a function`);
    }
  }

  if (tasks.length === 0) {
    return [];
  }

  const results = new Array(tasks.length);
  let nextIndex = 0;
  let stopped = false;

  async function worker() {
    while (!stopped) {
      const index = nextIndex;
      if (index >= tasks.length) {
        return;
      }

      nextIndex += 1;

      try {
        results[index] = await tasks[index]();
      } catch (error) {
        stopped = true;
        throw error;
      }
    }
  }

  const workerCount = Math.min(concurrency, tasks.length);
  const workers = Array.from({ length: workerCount }, () => worker());

  await Promise.all(workers);
  return results;
}

const sleep = (milliseconds) =>
  new Promise((resolve) => setTimeout(resolve, milliseconds));

// 三组以上可直接运行的验证；异步题使用独立执行块。
(async () => {
  console.log(await asyncPool([], 2)); // 预期：[]
  console.log(await asyncPool([() => 1, () => 2, () => 3], 2)); // 预期：[1, 2, 3]

  const completionOrder = [];
  const orderedResult = await asyncPool(
    [
      async () => {
        await sleep(30);
        completionOrder.push("A");
        return "A";
      },
      async () => {
        await sleep(10);
        completionOrder.push("B");
        return "B";
      },
      async () => {
        await sleep(20);
        completionOrder.push("C");
        return "C";
      },
    ],
    2,
  );

  console.log(completionOrder); // 预期：["B", "A", "C"]
  console.log(orderedResult); // 预期：["A", "B", "C"]

  let active = 0;
  let maxActive = 0;
  const concurrencyResult = await asyncPool(
    Array.from({ length: 9 }, (_, index) => async () => {
      active += 1;
      maxActive = Math.max(maxActive, active);
      await sleep(5);
      active -= 1;
      return index;
    }),
    3,
  );

  console.log(concurrencyResult); // 预期：[0, 1, 2, 3, 4, 5, 6, 7, 8]
  console.log(maxActive); // 预期：3

  const started = [];
  const expectedError = new Error("failed");
  console.log(await Promise.resolve(asyncPool(
      [
        async () => {
          started.push(0);
          await sleep(5);
          throw expectedError;
        },
        async () => {
          started.push(1);
          await sleep(30);
          return 1;
        },
        () => {
          started.push(2);
          return 2;
        },
      ],
      2,
    )).then(() => false, () => true)); // 预期：true（拒绝）

  await sleep(40);
  console.log(started); // 预期：[0, 1]

  let invalidInputStarted = false;
  console.log(await Promise.resolve(asyncPool(
      [
        () => {
          invalidInputStarted = true;
          return 1;
        },
        null,
      ],
      1,
    )).then(() => false, () => true)); // 预期：true（拒绝）
  console.log(invalidInputStarted); // 预期：false

  console.log(await Promise.resolve(asyncPool([], 0)).then(() => false, () => true)); // 预期：true（拒绝）


})().catch(console.error);

复杂度分析

时间复杂度：O(n)，不计任务本身。调度器对 n 个任务各领取和写回一次。

空间复杂度：O(n+c)。结果数组保存 n 项，至多 c 个 Worker 同时运行。

下载原始 JS 文件