直接用 ES6 class 构建一个支持高并发异步任务调度的流式数据清洗管道基类？听起来很唬人，但拆开来看，核心要点包括：阶段管理、并发控制、分批处理、错误隔离及钩子扩展。JavaScript 的单线程特性使其没有原生多线程并发能力，这里的“高并发”实际上指的是高吞吐、低延迟、可并行调度的异步流水线。实现上依赖 Promise 链、任务队列、并发数限制（比如 Promise.allSettled 配合信号量）以及可插拔的处理阶段设计。下面我们一步步拆解。
1. 定义管道基类：支持阶段注册与顺序执行
基类并不直接执行清洗任务，而是管理阶段（stages）、输入源（source）、输出目标（sink）及调度策略。每个 stage 是一个异步函数，接收数据并返回清洗后的结果，也可通过 reject 抛出错误。关键设计点包括：
- 构造函数接受可选的 maxConcurrency 参数（默认值为 3），用于限制同时运行的 stage 实例数量。
- 通过
Array.push()累积 stage，确保按注册顺序执行。 - 所有 stage 必须返回 Promise，统一采用
async/await或Promise.then进行封装。
示例代码：
class DataPipeline {
constructor(maxConcurrency = 3) {
this.stages = [];
this.maxConcurrency = maxConcurrency;
}
use(stageFn) {
if (typeof stageFn !== 'function') throw new TypeError('Stage must be a function');
this.stages.push(stageFn);
return this;
}
}2. 实现流式调度：按批+限流+错误隔离
清洗管道不应一次性加载所有数据——否则内存会溢出，同时也不能让一个失败的 stage 阻塞整个流程。推荐采用“分批处理 + 并发控制 + 失败跳过”的模式。
核心方法 process(items) 应实现以下功能：
- 将输入数组拆分为大小为 maxConcurrency 的批次。
- 对每一批调用
Promise.allSettled()，确保同一批次内的 stage 并行执行但互不干扰。 - 每个 stage 调用时包裹 try/catch，失败时不中断后续 stage，仅记录错误或打标记。
- 返回结构化结果：
{ data: cleanedItems[], errors: [] }。
示例片段：
async process(items) {
const results = [];
const errors = [];
const batches = this.#chunk(items, this.maxConcurrency);
for (const batch of batches) {
const settled = await Promise.allSettled(
batch.map(item => this.#runStages(item))
);
settled.forEach(r => {
if (r.status === 'fulfilled') results.push(r.value);
else errors.push(r.reason);
});
}
return { data: results, errors };
}
#runStages(item) {
return this.stages.reduce((acc, stage) => acc.then(data => stage(data)), Promise.resolve(item));
}3. 支持异步清洗阶段：每个 stage 可含 I/O 或计算
清洗阶段本身必须支持异步操作，例如去重查库、调用外部 API 校验手机号、格式化时间戳等场景。正确写法是返回 Promise：
const validatePhone = async (record) => {
const res = await fetch(`/api/validate?phone=${record.phone}`);
if (!res.ok) throw new Error(`API failed: ${res.status}`);
const valid = await res.json();
return { ...record, isValid: valid };
};错误写法是同步阻塞且缺乏错误处理：
// ❌ 不要这样：
const badStage = (item) => {
JSON.parse(item.raw); // 同步抛错会中断整个 pipeline
return item;
};4. 扩展性设计：支持中间件式钩子与生命周期
实际清洗流程往往需要日志记录、指标上报、超时控制、重试机制等。可以在基类中预留以下钩子：
- onStageStart(stageName, item)：在 stage 开始前触发。
- onStageError(stageName, item, error)：在 stage 报错时触发。
- onBatchComplete(batchIndex, resultCount)：每批处理完成后触发。
这些钩子默认实现为空函数，子类可以 override 或通过 options 注入，不影响其他功能。
