用JS实现多个任务并行执行的队列

使用场景

今天在开发时，碰到了这么一个需求：需要上传多个视频文件；每个文件的上传在新线程内执行；同时最多有三个上传任务在执行；当一个任务执行完毕，执行下一个任务。这个需求就是并行执行队列内的任务。

为了能够满足要求，需要知道可以同时执行的任务数、任务队列、正在执行的任务。当一个任务执行完毕后，需要分配下一个执行的任务。

代码实现

首先定义一个类：

class Queue {
  constructor(workerLen) {
    this.workerLen = workerLen ?? 3;         // 同时执行的任务数
    this.list = [];                          // 任务队列
    this.worker = new Array(this.workerLen); // 正在执行的任务
  }

  /**
   * 执行一个任务
   * @param { number } index
   * @param { Function } fn - 执行的函数
   * @param { Array } args - 传递给执行函数的参数
   */
  *executionFunc(index, fn, ...args) {
    const _this = this;

    yield fn.call(...args)
      .then(function() {
        // 任务执行完毕后，再次分配任务并执行任务
        _this.worker[index] = undefined;
        _this.run();
      });
  }

  /**
   * 添加到任务队列
   * @param { Array<any[]> } list: 任务队列
   */
  addList(list) {
    for (const item of list) {
      this.list.unshift(item);
    }
  }

  // 分配并执行任务
  run() {
    const runIndex = [];

    for (let i = 0; i < this.workerLen; i++) {
      const len = this.list.length;

      if (!this.worker[i] && len > 0) {
        // 需要执行的任务
        this.worker[i] = this.executionFunc(i, ...this.list[len - 1]);

        runIndex.push(i);

        // 从任务队列内删除任务
        this.list.pop();
      }
    }

    // 执行任务
    for (const index of runIndex) {
      this.worker[index].next();
    }
  }
}

定义一个延迟执行的异步函数，并执行任务，测试代码是否满足需求：

// 延迟执行的函数
function sleep(id, time) {
  console.log('开始id', id);

  return new Promise((resolve) => {
    setTimeout(() => {
      console.log('结束id', id);
      resolve();
    }, time * 1000);
  });
}

const queue = new Queue();

// 添加到任务队列
queue.addList([
  [sleep, undefined, '0001', 3],
  [sleep, undefined, '0002', 5],
  [sleep, undefined, '0003', 8],
  [sleep, undefined, '0004', 1],
  [sleep, undefined, '0005', 12],
  [sleep, undefined, '0006', 8],
  [sleep, undefined, '0007', 2],
  [sleep, undefined, '0008', 10]
]);

// 执行任务
queue.run();

在控制台上会输出：

开始id 0001
开始id 0002
开始id 0003
结束id 0001
开始id 0004
结束id 0004
开始id 0005
结束id 0002
开始id 0006
结束id 0003
开始id 0007
结束id 0007
开始id 0008
结束id 0006
结束id 0005
结束id 0008

可以看到，开始时执行了三个任务；每当有任务执行完毕，就会执行下一个任务。这样就满足了并行执行队列内的任务的需求。

代码演示

共有6个任务，每个任务过一定时间后完成。任务执行完成后执行下一个任务，最多有3个任务在执行。

任务ID	执行时间（s）	执行状态
1	3	未执行
2	9	未执行
3	12	未执行
4	5	未执行
5	7	未执行
6	1	未执行

已完成任务：无