跳到主内容
创建时间：最后更新：作者：白川

并行执行任务并汇总结果

在限制并发的条件下检查多个文件，保留身份与顺序，区分部分失败和整体取消。

并行执行任务并汇总结果

多个文件的只读检查互不依赖时，没有必要等待第一个完成才开始第二个。本篇同时检查几份文件，保留每份结果的身份，并让一个文件失败不掩盖其他成功结果。

沿用工作流项目与 src/lesson.ts。本篇不调用模型。另建 fixtures/other.ts：

typescript
export function size(items: unknown[]) {
  return items.length
}

先判断能否并行

读取不同文件通常可以独立进行。两个步骤同时修改同一个文件，则可能互相覆盖，不能因为都返回 Promise 就默认适合并发。

并行还需要限制数量：同时发起所有模型或工具调用，可能超过配额或占满资源。这里把“最多同时执行多少个”和“总共需要处理多少个”分开。

写一个带身份的并发执行器

创建 src/parallel.ts：

typescript
export type ItemResult<T> = { ok: true; value: T } | { ok: false; error: string }

export async function mapLimited<T, R>(
  items: readonly T[],
  limit: number,
  signal: AbortSignal,
  execute: (item: T, signal: AbortSignal) => Promise<R>,
): Promise<ItemResult<R>[]> {
  if (!Number.isInteger(limit) || limit < 1) {
    throw new Error("并发数必须是正整数")
  }
  const results: ItemResult<R>[] = new Array(items.length)
  let cursor = 0
  async function worker() {
    while (cursor < items.length) {
      signal.throwIfAborted()
      const index = cursor++
      try {
        results[index] = { ok: true, value: await execute(items[index], signal) }
      } catch (error) {
        if (signal.aborted) {
          throw error
        }
        results[index] = {
          ok: false,
          error: error instanceof Error ? error.message : String(error),
        }
      }
    }
  }
  // 等所有已启动 worker 收尾，再向调用者报告取消。
  await Promise.allSettled(Array.from({ length: Math.min(limit, items.length) }, () => worker()))
  signal.throwIfAborted()
  return results
}

结果按原始输入索引保存，所以完成先后不会改变文件对应关系。每个执行失败转为一个明确的结果，调用者可以选择继续汇总或把整个业务判为失败。

Promise.allSettled() 在这里用于等待已经启动的 worker 收尾。取消后不再领取新条目，但不能强制终止一个完全不响应信号的第三方操作；执行器必须继续传递 signal。

同时检查三份文件

创建 src/parallel-main.ts：

typescript
import { readFile } from "node:fs/promises"
import { extractFunction } from "./lesson.js"
import { mapLimited } from "./parallel.js"

const files = ["sample.ts", "other.ts", "missing.ts"] as const
const signal = AbortSignal.timeout(10_000)
const results = await mapLimited(files, 2, signal, async (file, itemSignal) => {
  const source = await readFile(new URL(`../fixtures/${file}`, import.meta.url), {
    encoding: "utf8",
    signal: itemSignal,
  })
  return extractFunction(source)
})
console.log(results.map((result, index) => ({ file: files[index], ...result })))

运行：

bash
node --import tsx src/parallel-main.ts

这里故意不创建 missing.ts。预期两个文件成功、一个文件失败，结果仍按输入顺序展示。文件不存在是本例准备的局部失败，不是需要不断重试的临时网络问题。

汇总规则要由业务决定

结果数组不是“全部成功”的标志。对于代码检查报告，可以保留成功项并标记缺失项；对于必须检查全部文件才能发布的流程，则应在发现任何失败时阻止发布。

不要直接丢弃失败项只显示成功数量，也不要把并发结果的到达顺序当作任务顺序。若结果要交给模型汇总，应同时传递文件身份和失败信息，并控制内容体积。

取消与局部失败的区别

单个文件出错，本例继续其他文件；整个信号被中止，执行器等待已启动的工作收尾后报告取消，不再返回一份看似完整的成功列表。

对有副作用的任务，取消不回滚已经完成的写入。本例只使用读取，便于先学习并发状态。多个智能体之间的角色、隔离和交接另在多智能体专题展开。

本篇验证了并发上限、顺序、局部失败和中止收尾。下一篇让检查结果驱动有限反馈循环。

参考资料