如何使用ES6 class构建支持高并发异步任务调度的流式数据清洗管道基类-js教程-PHP中文网

datapipeline类实现高并发异步任务调度的流式数据清洗管道，支持阶段注册、并发限制、分批处理、错误隔离及钩子扩展。

如何使用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. 实现流式调度：按批+限流+错误隔离

清洗管道不能一次性 load 所有数据（内存溢出），也不应让一个失败 stage 阻塞整条流。推荐使用“分批处理 + 并发控制 + 失败跳过”模式。

核心方法 process(items) 应：

Web前端Promise教程从入门到精通课件

下载

将输入数组切分为大小为 maxConcurrency 的批次
对每一批调用 Promise.allSettled()，确保单批内 stage 并行但互不干扰
每个 stage 调用包裹 try/catch，失败时不中断后续 stage，记录 error 或打标记
返回结构化结果：{ 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));
}