Strapi 异步工具集详解：@strapi/utils 中 async.map / async.reduce / async.pipe 的实现原理与实战

本文基于 Strapi 核心文档 Async utils functions，系统讲解 @strapi/utils 包中 async 命名空间提供的三个工具函数——map、reduce 与 pipe 的 API 用法、源码实现与测试行为。读完本文，你将能够在插件与 Core 服务代码中正确使用异步数组遍历与函数组合，理解其底层依赖（p-map、lodash/fp）带来的并发与柯里化特性，并了解 Strapi 官方推荐的扩展方向（filterAsync、retryAsync、timeoutAsync 等）。

一、async utils 的定位：Strapi 的 Promise 工具层

Strapi 官方文档对 async utils 的定义是：这一组函数专门用于处理与 Promise 相关的异步逻辑（原文："Async utils are grouping all function that interact with async stuff like Promises"）。

从源码结构看，async 命名空间在 packages/core/utils/src/index.ts 中通过命名空间导出对外发布：

export * as async from './async';

其完整实现集中在 packages/core/utils/src/async.ts，全文不足 40 行，仅依赖两个成熟库：

import pMap from 'p-map';
import { curry } from 'lodash/fp';

• map 是对 p-map（版本锁定为 4.0.0，见 packages/core/utils/package.json 的 dependencies）的柯里化包装；
• pipe 与 reduce 则是手写的轻量实现。

该包通过 @strapi/utils 包名发布，仓库内所有 Core 包（admin、content-manager、content-releases 等）均可直接 import { async } from '@strapi/utils' 使用。运行环境前提：@strapi/utils 的 engines 声明要求 Node >=20.0.0 <=26.x.x。

官方文档同时给出了使用准则：

When to use：Every time the code has to act with promises and iterate other them, an async utils function should be used.

Should I add my function here?：Any util function that manipulates promises can be included in this utils section.（即：任何操作 Promise 的工具函数都可以归入该模块，但若函数数量膨胀，需评估是否迁移到专用库，见第五节。）

二、async.map：支持并发控制的异步 Array.prototype.map

2.1 用法

官方文档给出的示例（非柯里形式，直接传数组与迭代器）：

import { async } from '@strapi/utils';

const input = [1, 2, 3];

const output = await async.map(input, async (item) => {
  return item * 2;
});

console.log(output); // [2, 4, 6]

由于 map 被 lodash/fp 的 curry 包裹，它同样支持柯里化的部分应用形式——先绑定数组，再传入迭代器（甚至并发选项）。测试文件 packages/core/utils/src/tests/async.test.ts 正是采用这种写法：

const mapFunc = map(numberPromiseArray);
const result = await mapFunc((number) => number + 1);

2.2 实现：curry(pMap)

源码只有一行：

export const map = curry(pMap);

因此它完整继承 p-map 的语义：

• 迭代器返回 Promise 或普通值均可，输入数组中混合已 resolve 的 Promise 与同步值也能正常工作（测试用例 "Should work with mix of promises and values" 验证了 [1, Promise.resolve(2)] 映射结果为 [2, 3]）；
• 任意一个元素 reject 或迭代器抛错，整个 map 立即 reject，错误向上传播（测试用例验证了迭代器抛 'test' 与输入含 Promise.reject(new Error('input')) 两种场景）；
• 支持 options.concurrency 限制并发数。测试用例 "Should resolve elements two at a time" 用 6 个元素、每个任务延迟 20ms 的场景断言：传入 { concurrency: 2 } 后，峰值并发 maxOperations 恒等于 2，且结果数组保持原顺序 [1, 2, 3, 4, 5, 6]。

这与 Promise.all(input.map(fn)) 的朴素写法相比，async.map 的价值在于可显式控制并发度——对批量数据库查询、远程 API 调用等场景，可以防止一次性发出过多并发请求。

2.3 Strapi 中的真实用法

Core 代码中大量使用 async.map 对查询结果做逐条异步处理，例如 content-manager 的文档元数据格式化 packages/core/content-manager/server/src/controllers/utils/metadata.ts：

availableLocales = await async.map(
  availableLocales,
  async (localeDocument: AvailableLocaleDocument) => metadataSanitizer(localeDocument)
);

以及 content-types.ts 控制器 中先 async.map 逐条处理、再串联 async.pipe(permissionChecker.sanitizeOutput, setStatus) 的写法；content-releases 的数据库迁移 packages/core/content-releases/server/src/migrations/index.ts 也用 async.map(releasesWithoutStatus, ...) 批量改写发布状态。

三、async.reduce：顺序执行的异步归约

3.1 用法

reduce 是 Array.prototype.reduce 的异步版本，且采用强制柯里化的签名——第一个参数必须是数组，返回一个接受 (iteratee, initialValue?) 的函数：

import { async } from '@strapi/utils';
const input = [1, 2, 3];

const reducer = async.reduce(input);
const output = await reducer(async (accumulator, item) => {
  return accumulator + item;
}, 0);

console.log(output); // 6

3.2 实现：逐元素 await 的串行循环

packages/core/utils/src/async.ts 中的实现是朴素的 for 循环，语义为严格顺序执行（无并发，这与 map 形成对比）：

export const reduce =
  (mixedArray: any[]) =>
  async <T>(iteratee: AnyFunc, initialValue?: T) => {
    let acc = initialValue;
    for (let i = 0; i < mixedArray.length; i += 1) {
      acc = await iteratee(acc, await mixedArray[i], i);
    }
    return acc;
  };

几个关键行为，均可在 async.test.ts 中找到对应断言：

行为	测试用例	说明
支持初始值	"Should return an incremented number"	reduce([1,2])((p, c) => p + c, 10) 得 13
支持省略初始值	"Should work without initial value"	initialValue 类型为 T | undefined，首次迭代时累积器为 undefined，迭代器需自行兜底
支持 Promise 与同步值混合	"Should work with mix of promises and values"	每个元素先 await mixedArray[i] 再交给迭代器
回调参数含索引	实现签名	第三个参数 i 等价于原生 reduce 的 index
错误传播	"Should throw an error…" 两例	迭代器抛错、或输入含 rejected Promise，均使整体 reject

串行语义使其适用于前一步结果依赖上一步的累积场景，如顺序构建上下文、按索引更新阶段状态等。Strapi 内部示例：packages/core/content-releases/server/src/services/release-action.ts 中 await async.reduce(contentTypeUids)(...) 逐个内容类型构建模型映射；packages/core/review-workflows/server/src/services/stages.ts 中 await async.reduce(stagesList)(async (_, stage, idx) => ...) 按索引处理阶段列表。

四、async.pipe：异步函数的顺序组合

4.1 用法

pipe 用于组合异步函数：接收一组函数，返回一个新函数，按顺序依次应用（前一个的输出作为后一个的输入）：

import { async } from '@strapi/utils';

async function addOne(input: number): Promise<number> {
  return input + 1;
}

async function double(input: number): Promise<number> {
  return input * 2;
}

const addOneAndDouble = async.pipe(addOne, double);

const output = await addOneAndDouble(3);

console.log(output); // (3 + 1) * 2 = 8

4.2 实现：运行时循环 + 编译期类型推导

async.ts 中 pipe 的运行时实现是"首函数带参调用 + 其余函数逐个 await 串联"：

export function pipe<T extends AnyFunc[]>(...fns: PipeReturn<T> extends never ? never : T) {
  const [firstFn, ...fnRest] = fns;

  return (async (...args: any[]) => {
    let res = await firstFn.apply(firstFn, args);

    for (let i = 0; i < fnRest.length; i += 1) {
      res = await fnResti;
    }

    return res;
  }) as PipedFunc<T>;
}

配合两个类型工具完成端到端的类型推导：

type MakeProm<T> = Promise<T extends PromiseLike<infer I> ? I : T>;

type PipedFunc<T extends AnyFunc[]> =
  PipeReturn<T> extends never ? never : (...args: Parameters<T[0]>) => PipeReturn<T>;

type PipeReturn<F extends AnyFunc[]> = MakeProm<ReturnType<F[0]>>;

这意味着：

• 组合函数的入参类型由第一个函数的 Parameters 决定，返回值是 Promise<第一个函数的返回类型（解包 Promise 后）>；
• 中间函数可以是同步函数、返回 Promise 的异步函数混用，await 对两者都成立。测试用例 "Should pipe several functions" 验证了 [同步 n*n, 异步 n*PI, 同步 Math.round] 的混合管道，circleArea(50) 得到 7854。

一个值得注意的细节：pipe 只以第一个函数的返回类型标注结果，因此如果中间函数改变了类型（如 number → string），从源码结构看 TypeScript 层面不会报错但语义上是"宽松"的，团队内使用管道时应保证链上类型一致。

4.3 Strapi 中的真实用法

async.pipe 在 Core 中主要用于把数据库查询、序列化、权限清洗串成管道。典型如 admin 服务启动时的 API Token 权限同步 packages/core/admin/server/src/bootstrap.ts：

const permissionsInDB = await async.pipe(
  strapi.db.query('admin::api-token-permission').findMany,
  map('action')
)();

这里管道的第一环是数据库 findMany 查询函数，第二环是 lodash 的 map('action') 提取动作名，管道函数在定义后以 () 立即调用。content-manager 的 sanitize.ts 与 validate.ts 则用 async.pipe(sanitizeFields, sanitizeInput, ...) 将多步输入清洗/校验函数组合成单一管道，控制器中直接 ctx.body = await async.pipe(...)(...)。这种模式使"查询 → 逐条转换 → 序列化"的步骤显式化，且每一环都是可单独测试的纯函数。

五、何时使用 async utils，以及未来的扩展方向

5.1 使用准则（继承官方文档）

• When to use：只要代码需要对 Promise 进行遍历/归约/组合，应优先使用 async utils，而不是手写 for...of + await 或 Promise.all(arr.map(fn))。前者丢失了 Strapi 内部统一的并发控制与错误传播约定，后者则无法限制并发；
• Should I add my function here：任何操作 Promise 的工具函数都可归入 packages/core/utils/src/async.ts，但文档同时提醒：如果 async 节函数数量持续膨胀，应考虑直接迁移到专用异步库（文档点名的候选是 asyncjs 系列库），避免自维护成本超过收益。

5.2 官方文档列出的候选扩展函数

文档 "Potential improvements" 一节明确列出了尚未实现、但被认为值得加入的方向：

1. 其余 Array.prototype 方法：filterAsync、someAsync、everyAsync、findAsync、findIndexAsync、flatMapAsync；
2. retryAsync：失败后按指定次数重试的异步操作包装器——输入一个异步操作与重试次数，成功则返回结果，重试耗尽则抛出错误；
3. timeoutAsync：为异步操作附加超时——输入操作与超时时长，超时前完成则返回结果，否则抛出超时错误。

这三类函数分别补齐了"谓词遍历""容错""时限控制"三个当前 map/reduce/pipe 不覆盖的维度，可作为评估该模块演进方向的参考。

六、参考文件索引

内容	相对路径
本文所依据的设计文档	docs/docs/docs/01-core/utils/async.md
async 工具函数实现	packages/core/utils/src/async.ts
@strapi/utils 命名空间导出	packages/core/utils/src/index.ts
单元测试（map/reduce/pipe 行为断言）	packages/core/utils/src/tests/async.test.ts
包声明与 p-map/lodash 依赖版本	packages/core/utils/package.json
async.pipe 真实用例（admin 启动）	packages/core/admin/server/src/bootstrap.ts
async.map 真实用例（元数据清洗）	packages/core/content-manager/server/src/controllers/utils/metadata.ts
async.reduce 真实用例（发布动作/阶段服务）	packages/core/content-releases/server/src/services/release-action.ts、packages/core/review-workflows/server/src/services/stages.ts