muliminty note

队列来控制并发任务

7分钟阅读

在JavaScript中，可以通过以下步骤实现一个队列来控制并发任务的顺序执行：

实现思路

  1. 队列结构：使用数组存储待执行任务。
  2. 并发控制：设置最大并发数，跟踪当前运行的任务数。
  3. 任务调度：当有任务完成时，从队列中取出下一个任务执行，确保按添加顺序启动。

代码实现

class ConcurrentQueue {
  constructor(concurrency = 1) {
    this.concurrency = concurrency; // 最大并发数
    this.running = 0; // 当前运行中的任务数
    this.queue = []; // 待执行的任务队列
  }
 
  // 添加任务到队列
  add(task) {
    this.queue.push(task);
    this.next();
  }
 
  // 执行下一个任务
  next() {
    // 当未达到并发上限且队列中有任务时
    while (this.running < this.concurrency && this.queue.length > 0) {
      const task = this.queue.shift(); // 取出队列中的第一个任务
      this.running++;
      task().finally(() => {
        this.running--;
        this.next(); // 当前任务完成后继续调度
      });
    }
  }
}

使用示例

// 示例：模拟异步任务
const delay = (ms, msg) => 
  new Promise(resolve => setTimeout(() => {
    console.log(msg);
    resolve();
  }, ms));
 
// 创建并发数为2的队列
const queue = new ConcurrentQueue(2);
 
// 添加任务（按顺序启动）
queue.add(() => delay(1000, '任务1完成'));
queue.add(() => delay(500, '任务2完成'));
queue.add(() => delay(800, '任务3完成'));
queue.add(() => delay(200, '任务4完成'));
 
/* 输出顺序：
   "任务2完成" (500ms后)
   "任务1完成" (1000ms后)
   "任务4完成" (1200ms后，任务1完成后启动)
   "任务3完成" (1300ms后，任务2完成后启动)
*/

关键点解析

  1. 按序启动：通过shift()从队列头部取出任务，确保先添加的任务先执行。
  2. 并发控制：while循环保证任何时候运行的任务数不超过concurrency。
  3. 自动调度：每个任务完成后自动触发next()，继续执行后续任务。

适用场景

  • API请求限速（如避免触发反爬机制）
  • 文件批量处理（如控制同时打开的文件数）
  • 异步任务流水线（如按顺序处理但允许并行加速）

如果需要严格按完成顺序处理结果，需额外逻辑（如收集结果后排序），但此实现已满足基本并发控制需求。

要实现任务的严格顺序执行（即前一个任务完成后才能开始下一个），无论任务本身的执行时间长短，需要将并发数设置为 1，即改为串行队列。以下是修改后的实现：

class SerialQueue {
  constructor() {
    this.queue = [];     // 任务队列
    this.isRunning = false; // 当前是否有任务在执行
  }
 
  // 添加任务到队列
  add(task) {
    this.queue.push(task);
    this.run();
  }
 
  // 执行任务（严格串行）
  async run() {
    if (this.isRunning) return;
    this.isRunning = true;
 
    while (this.queue.length > 0) {
      const task = this.queue.shift(); // 按顺序取任务
      await task(); // 等待当前任务完成
    }
 
    this.isRunning = false;
  }
}

使用示例

const queue = new SerialQueue();
 
queue.add(() => delay(1000, '任务1完成'));
queue.add(() => delay(500, '任务2完成'));
queue.add(() => delay(800, '任务3完成'));
queue.add(() => delay(200, '任务4完成'));
 
/* 输出顺序（严格按添加顺序完成）：
   "任务1完成" (1000ms后)
   "任务2完成" (1500ms后)
   "任务3完成" (2300ms后)
   "任务4完成" (2500ms后)
*/

关键变化说明

  1. 去掉了并发控制：强制每次只执行一个任务，通过 await 等待当前任务完成。
  2. 执行逻辑简化：使用 while 循环逐个执行任务，确保绝对顺序。

如果你需要 **并发执行但按顺序处理结果

如果希望任务可以并行执行（如并发网络请求），但结果处理按顺序进行，可以使用以下方案：

class OrderedResultQueue {
  constructor() {
    this.queue = [];      // 任务队列
    this.nextOrder = 0;   // 期待的下一个任务序号
    this.results = new Map(); // 存储已完成任务的结果
  }
 
  // 添加任务（需自带序号）
  add(task, order) {
    this.queue.push({ task, order });
    this.run();
  }
 
  // 执行任务（可并行）
  async run() {
    // 所有任务并行执行
    const promises = this.queue.map(async ({ task, order }) => {
      const result = await task();
      this.results.set(order, result);
    });
 
    await Promise.all(promises);
    this.queue = [];
 
    // 按顺序输出结果
    while (this.results.has(this.nextOrder)) {
      console.log(this.results.get(this.nextOrder));
      this.results.delete(this.nextOrder);
      this.nextOrder++;
    }
  }
}

使用示例

const queue = new OrderedResultQueue();
 
// 添加任务时需指定顺序标识
queue.add(() => delay(1000, '任务1完成'), 1);
queue.add(() => delay(500, '任务2完成'), 2);
queue.add(() => delay(800, '任务3完成'), 3);
queue.add(() => delay(200, '任务4完成'), 4);
 
/* 输出顺序：
   "任务1完成" (1000ms后)
   "任务2完成" (1000ms后，但等待任务1完成后一起按顺序输出)
   "任务3完成"
   "任务4完成"
*/

方案对比

方案执行方式结果顺序适用场景
SerialQueue串行执行严格顺序必须前一个任务完成后才能开始下一个
OrderedResultQueue并行执行按序处理任务可并行但结果需要按顺序使用

根据你的实际需求选择合适的实现方式。

关系图谱