如何基于async/await实现指定并行度的函数队列并行处理？

看起来你正在尝试实现一个带并行度控制的异步任务队列——这种需求在处理大量异步操作（比如网络请求、文件处理）时非常常见。我先帮你梳理下现有代码的问题，再给出更健壮的实现方案。

现有代码的潜在问题 #

你当前的startChain逻辑存在一个明显的bug：当internalQueue为空时，pickUpNextTask会返回undefined，这时候调用.then(next)会直接抛出错误，因为undefined并没有then方法。另外，这种链式Promise的方式虽然能串行处理任务，但要实现多并行链的话，代码会变得繁琐，且不易跟踪活跃任务的数量。

推荐实现方案 #

我推荐用活跃任务计数的方式来控制并行度，这种方式更直观、健壮，也更容易处理错误和动态添加任务的场景。

类封装版本（适合长期维护/复用） #

class ParallelQueue {
  constructor(processFunc, concurrency) {
    this.processFunc = processFunc; // 每个任务要执行的函数
    this.concurrency = concurrency; // 最大并行度
    this.taskQueue = []; // 待处理的任务队列
    this.activeTasks = 0; // 当前正在运行的任务数
  }

  // 添加单个任务到队列
  addTask(taskData) {
    this.taskQueue.push(taskData);
    this._startProcessing();
  }

  // 批量添加任务
  addTasks(taskDataArray) {
    this.taskQueue.push(...taskDataArray);
    this._startProcessing();
  }

  // 核心处理逻辑
  async _startProcessing() {
    // 只有当活跃任务数小于并行度，且队列还有任务时才继续
    while (this.activeTasks < this.concurrency && this.taskQueue.length > 0) {
      const taskData = this.taskQueue.shift();
      this.activeTasks++;
      
      try {
        await this.processFunc(taskData);
        // 这里可以添加任务成功后的回调逻辑
      } catch (error) {
        // 处理任务执行失败的情况，比如日志记录、错误上报
        console.error(`任务处理失败:`, error);
      } finally {
        this.activeTasks--;
        // 当前任务完成后，递归触发下一轮处理
        this._startProcessing();
      }
    }
  }

  // 等待所有任务执行完成
  async waitForCompletion() {
    while (this.activeTasks > 0 || this.taskQueue.length > 0) {
      await new Promise(resolve => setTimeout(resolve, 50));
    }
  }
}

使用示例 #

// 定义你的任务处理函数
async function myTaskHandler(item) {
  console.log(`开始处理任务: ${item}`);
  // 模拟异步操作（比如API请求、文件读写）
  await new Promise(resolve => setTimeout(resolve, 1000));
  console.log(`完成任务: ${item}`);
  return item * 2;
}

// 初始化队列，设置并行度为3
const queue = new ParallelQueue(myTaskHandler, 3);

// 添加任务
queue.addTasks([1, 2, 3, 4, 5, 6, 7]);

// 等待所有任务完成后做后续操作
queue.waitForCompletion().then(() => {
  console.log("所有任务执行完毕！");
});

函数式版本（适合一次性使用） #

如果不需要长期复用，也可以用更简洁的函数式实现：

async function processWithConcurrency(items, processFunc, concurrency) {
  const taskQueue = [...items];
  let activeTaskCount = 0;
  const results = [];

  async function processNextTask() {
    if (taskQueue.length === 0) return;
    
    activeTaskCount++;
    const item = taskQueue.shift();
    
    try {
      const result = await processFunc(item);
      results.push(result);
    } catch (err) {
      console.error(`处理任务失败:`, item, err);
      results.push(null); // 可根据需求自定义错误结果
    } finally {
      activeTaskCount--;
      await processNextTask();
    }
  }

  // 启动指定数量的并行处理链
  const initialWorkers = Array.from({ length: concurrency }, processNextTask);
  await Promise.all(initialWorkers);
  
  return results;
}

函数式版本使用示例 #

const tasks = [10, 20, 30, 40, 50];
processWithConcurrency(tasks, myTaskHandler, 2).then(finalResults => {
  console.log("所有任务结果:", finalResults);
});

方案优势 #

• 精准控制并行度，不会超过设定的最大并发数
• 自动处理任务队列，新增任务后会自动启动处理
• 完善的错误处理，单个任务失败不会影响整个队列
• 支持等待所有任务完成，方便后续流程衔接

内容的提问来源于stack exchange，提问作者Moshe Shmukler