js 实现一个任务队列

实现基础任务队列结构

使用数组存储任务，通过push方法添加任务，shift方法取出任务执行：

class TaskQueue {
  constructor() {
    this.tasks = [];
    this.isRunning = false;
  }

  add(task) {
    this.tasks.push(task);
    if (!this.isRunning) this.run();
  }

  async run() {
    this.isRunning = true;
    while (this.tasks.length > 0) {
      const task = this.tasks.shift();
      try {
        await task();
      } catch (error) {
        console.error('Task error:', error);
      }
    }
    this.isRunning = false;
  }
}

添加并发控制功能

扩展基础队列实现最大并发数限制：

class ConcurrentQueue {
  constructor(concurrency = 1) {
    this.tasks = [];
    this.concurrency = concurrency;
    this.running = 0;
  }

  add(task) {
    this.tasks.push(task);
    this.next();
  }

  next() {
    while (this.running < this.concurrency && this.tasks.length) {
      const task = this.tasks.shift();
      this.running++;
      task().finally(() => {
        this.running--;
        this.next();
      });
    }
  }
}

实现优先级队列

通过优先级参数控制任务执行顺序：

class PriorityQueue {
  constructor() {
    this.highPriority = [];
    this.lowPriority = [];
    this.isRunning = false;
  }

  add(task, priority = 'low') {
    const queue = priority === 'high' ? this.highPriority : this.lowPriority;
    queue.push(task);
    if (!this.isRunning) this.run();
  }

  async run() {
    this.isRunning = true;
    while (this.highPriority.length || this.lowPriority.length) {
      const queue = this.highPriority.length ? this.highPriority : this.lowPriority;
      const task = queue.shift();
      try {
        await task();
      } catch (error) {
        console.error('Task error:', error);
      }
    }
    this.isRunning = false;
  }
}

带错误恢复的队列实现

增加错误重试机制和失败回调：

class ResilientQueue {
  constructor(maxRetries = 3) {
    this.tasks = [];
    this.maxRetries = maxRetries;
  }

  add(task, onError = null) {
    this.tasks.push({ task, retries: 0, onError });
    if (this.tasks.length === 1) this.run();
  }

  async run() {
    while (this.tasks.length) {
      const current = this.tasks[0];
      try {
        await current.task();
        this.tasks.shift();
      } catch (error) {
        current.retries++;
        if (current.retries >= this.maxRetries) {
          if (current.onError) current.onError(error);
          this.tasks.shift();
        }
      }
    }
  }
}

使用示例

基础队列使用方式：

const queue = new TaskQueue();
queue.add(() => new Promise(resolve => {
  setTimeout(() => {
    console.log('Task 1 done');
    resolve();
  }, 1000);
}));
queue.add(() => console.log('Task 2'));

并发队列示例：

const concurrent = new ConcurrentQueue(2);
for (let i = 1; i <= 5; i++) {
  concurrent.add(() => new Promise(resolve => {
    setTimeout(() => {
      console.log(`Task ${i} completed`);
      resolve();
    }, 1000);
  }));
}