灏天阁

Promise？异步并发任务控制器？

· Yin灏

前言

“实现一个控制并发数的任务队列 、实现一个异步并发任务控制器” 等，已经是非常经典的手写题目了，因为其中涉及 异步 和 并发 的内容，在正式开始实现之前我们先来简单了解一下它们的概念，毕竟只有知道为什么才能更好的实现，而不是单纯的去记忆。

494CDD6B.jpg

异步 & 并发

异步

单线程的 JavaScript

我们都知道 默认情况 下 JavaScript 是 单线程 的，又或者说 JavaScript 只在一个线程上运行。

【注意】JavaScript 虽然只在一个线程上运行，但不表示 JavaScript 引擎只有一个线程，实际上，JavaScript 引擎有多个线程，单个脚本只能在一个线程上运行（即 主线程），其他线程都是在后台配合

而 单线程 就意味着，所有任务需要 排队，前一个任务结束，才会执行后一个任务，如果前一个任务耗时很长，后一个任务就不得不一直等着。

JavaScript 异步的产生

如果排队是因为计算量大，CPU 处理不过来，这时候也算合理，但很多时候 CPU 是空闲的，是因为 IO 设备（输入/输出设备）很慢（比如 Ajax 操作从网络读取数据），CPU 不得不等着结果返回，才能继续往下执行。

JavaScript 语言的设计者意识到，这时主线程完全可以不管 IO 设备，挂起处于等待中的任务，先运行排在后面的任务，等到 IO 设备返回了结果，再回过头，把挂起的任务继续执行下去。

在 JavaScript 为了更好的处理异步问题，我们通常都会选择使用 Promise 或 async/await。

并发

早期计算机的 CPU 是 单核的，一个 CPU 在 同一时间 只能执行 一个进程/线程，当系统中有 多个进程/线程 等待执行时，CPU 只能执行完一个再执行下一个。

而所谓的 并发，指在同一时刻只能有一条 进程指令 执行，但多个 进程指令 被快速的 交替执行，那么在宏观上看就是多个进程同时执行的效果，但在微观上并不是同时执行的，只是把时间分成若干段，使多个进程快速交替的执行。

image.png

实现异步并发任务控制器

通过上述内容我们已经知道了 异步 和 并发 的基本概念，现在开始具体实现吧！

题目如下：

假设现在要发送多个请求，但要实现并发控制，即可以通过一个 limit 控制并发数，当任务数量超过对应的 limit 限制的并发数时，后续的任务需要延迟到 正在执行中 的任务执行完后 再执行，并且需要支持动态添加 额外的异步任务，同时当 最后一个任务 执行完成，需要执行对应的 callback 函数。

生成任务集合

// 生成用于测试的任务集合
const tasks = new Array(10).fill(0).map((v, i) => {
  return function task() {
    return new Promise((resolve) => {
      setTimeout(() => {
        resolve(i + 1);
      }, i * 1000);
    });
  };
});

方式一：并发控制函数 concurrencyControl

核心思路

  • 通过循环执行当前队列头部的任务
  • 当前队列头部任务执行完毕
    • 若是最后一个任务，则执行 callback
    • 否则，继续执行 下一个队头任务
// 并发控制函数
function concurrencyControl(tasks, limit, callback) {
  const queue = tasks.slice(); // 当前执行的任务队列
  let count = 0; // 已完成的任务数量

  const runTask = () => {
    while (limit) {
      limit--;

      if (queue.length) {
        const task = queue.shift(); // 取出当前队头任务

        task().then(() => {
          limit++;
          count++;

          if (count === tasks.length) {
            // 最后一个任务
            callback(); // 执行回调函数
          } else {
            runTask(); // 继续执行下一个任务
          }
        });
      }
    }
  };

  return runTask;
}

// 测试代码
const sendRequest = concurrencyControl(tasks, 3, (taskId) => {
  console.log(`task ${taskId} finish！`);
});

sendRequest();

不同时间的任务：

1.gif

相同时间的任务：

1.gif

方式二：并发控制器 ConcurrencyControl

方式一 虽然能够简单的完成自动化的并发控制，但是不支持 动态添加任务 的要求，这就意味着要 保持状态 了，并且如果当前执行的 promise 任务状态为 rejected 时就无法执行完全部的任务，因为 task().then 对应的 onreject 的回调没有被提供，下面我们就可以通过一个 ConcurrencyControl 类来实现。

核心思路

  • 将原本使用到的变量，转换成对应的实例属性
  • 新增 addTask() 方法用于动态添加任务，并且在其内部自动启动任务执行
  • task().then 替换为 task().finally，目的是当对应的 promise 任务为 reject 状态时仍能够执行
class ConcurrencyControl {
  constructor(tasks, limit, callback) {
    this.queue = tasks.slice(); // 当前执行的任务队列
    this.tasks = tasks; // 原始任务集合
    this.count = 0; // 已完成的任务数量
    this.limit = limit;
    this.callback = callback;
  }

  runTask() {
    while (this.limit) {
      this.limit--;

      if (this.queue.length) {
        const task = this.queue.shift(); // 取出队头任务

        task().finally(() => {
          this.limit++;
          this.count++;

          if (this.count === this.tasks.length) {
            // 最后一个任务
            this.callback(); // 执行回调函数
          } else {
            this.runTask(); // 继续执行下一个任务
          }
        });
      }
    }
  }

  addTask(task) {
    // 同步添加任务
    this.queue.push(task);
    this.tasks.push(task);
    // 当直接调用 addTask 也可直接执行
    this.runTask();
  }
}

// 测试代码
const Control = new ConcurrencyControl(tasks, 3, () => {
  console.log(`task all finish！`);
});

// 执行队列任务
Control.runTask();

// 添加新任务
Control.addTask(function task() {
  return new Promise((resolve) => {
    setTimeout(() => {
      console.log(`task ${Control.tasks.length} finish！`);
      resolve(Control.tasks.length);
    }, Control.tasks.length * 200);
  });
});

方式三：优化 并发控制器 ConcurrencyControl

核心思路

  • 优化掉 this.count 计数，通过 this.queue.size 来代替
  • 优化掉 this.addTask() 方法中的 this.queue.push(task)，通过 this.tasks 的变化来自动影响 this.queue 队列
  • 优化掉 this.limit ++/–，通过 this.queue.size < this.limit 来替换
class ConcurrencyControl {
  constructor(tasks, limit, callback) {
    this.tasks = tasks.slice(); // 浅拷贝，避免修改原数据
    this.queue = new Set(); // 任务队列
    this.limit = limit; // 最大并发数
    this.callback = callback; // 回调
  }

  runTask() {
    // 边界判断
    if (this.tasks.length == 0) return;

    // 当任务队列有剩余，继续添加任务
    while (this.queue.size < this.limit) {
      const task = this.tasks.shift(); // 取出队头任务

      this.queue.add(task); // 往队列中添加当前执行的任务

      task().finally(() => {
        this.queue.delete(task); // 当前任务执行完毕，从队列中删除改任务

        if (this.queue.size == 0) {
          this.callback(); // 执行回调函数
        } else {
          this.runTask(); // 继续执行下一个任务
        }
      });
    }
  }

  addTask(task) {
    // 同步添加任务
    this.tasks.push(task);
    // 当直接调用 addTask 也可直接执行
    this.runTask();
  }
}

// 测试代码
const Control = new ConcurrencyControl(tasks, 3, () => {
  console.log(`task all finish！`);
});

Control.runTask(); // 执行队列任务

Control.addTask(function task() {
  // 添加新任务
  return new Promise((resolve) => {
    setTimeout(() => {
      console.log(`task 9999 finish！`);
      resolve(999);
    }, 100);
  });
});

- Book Lists -