ES6 class构建高并发异步任务调度流式数据清洗管道基类

直接用 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 注入，不影响其他功能。