K Kiara's Tech Planet Tech Planet
Interview / JavaScript

手写异步任务调度器（Async Scheduler）

JavaScript 手写题 异步 面试

手写异步任务调度器（Async Scheduler）

面试高频手写题。考察点：并发控制、Promise 链、队列调度思想。

🎯 题目要求

实现一个 Scheduler 类，限制同时运行的异步任务最多为 N 个。超出的任务排队等候，前面的任务完成后自动拉取下一个执行。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
const scheduler = new Scheduler(2); // 最大并发 2

const timeout = (ms) => new Promise(resolve => setTimeout(resolve, ms));

async function addTask(delay, name) {
  await scheduler.add(() => timeout(delay));
  console.log(name);
}

addTask(1000, 'A'); // 1000ms 后输出
addTask(500,  'B'); //  500ms 后输出
addTask(300,  'C'); //  300ms 后输出（等 B 完成后立刻执行，共 800ms）
addTask(400,  'D'); //  400ms 后输出（等 A 完成后立刻执行，共 1400ms）

// 输出顺序：B → C → A → D

💡 核心思路

1
2
3
4
5
6
时间轴：
0ms     → A(1000ms)、B(500ms) 同时开始（占满 2 个槽位）
500ms   → B 完成，输出 B；立刻拉取 C(300ms) 开始
800ms   → C 完成，输出 C；D 进入等待（A 还没完）
1000ms  → A 完成，输出 A；立刻拉取 D(400ms) 开始
1400ms  → D 完成，输出 D

关键：不是「等所有任务完成再批量开下一批」，而是「一个坑位空出来就立刻填」。

✍️ 实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class Scheduler {
  constructor(max) {
    this.max     = max;  // 最大并发数
    this.running = 0;    // 当前正在运行的任务数
    this.queue   = [];   // 等待队列
  }

  /**
   * 添加一个任务
   * @param {() => Promise} promiseCreator - 返回 Promise 的函数（不是 Promise 本身！）
   * @returns {Promise} 任务完成时 resolve
   */
  add(promiseCreator) {
    return new Promise((resolve, reject) => {
      // 把「任务」和它的 resolve/reject 一起入队
      this.queue.push({ promiseCreator, resolve, reject });
      this._run();
    });
  }

  _run() {
    // 有空位 && 队列里有待执行的任务 → 取出来跑
    while (this.running < this.max && this.queue.length > 0) {
      const { promiseCreator, resolve, reject } = this.queue.shift();
      this.running++;

      promiseCreator()
        .then(resolve, reject)
        .finally(() => {
          this.running--;
          this._run(); // 任务完成 → 尝试拉取下一个
        });
    }
  }
}

为什么传的是「函数」而不是「Promise」？

1
2
3
4
5
// ❌ 错误：Promise 一 new 就立刻开始执行了，根本没法调度
scheduler.add(fetch('/api/data'));

// ✅ 正确：传一个「工厂函数」，由调度器决定什么时候调用它
scheduler.add(() => fetch('/api/data'));

🔍 逐行追踪（以上面的例子为例）

时间 事件 running queue
0ms 添加 A、B → 直接执行 2 []
0ms 添加 C → 已满，入队 2 [C]
0ms 添加 D → 已满，入队 2 [C, D]
500ms B 完成，running-- → _run() 拉取 C 2 [D]
800ms C 完成，running-- → _run() 尝试拉取 D，但 A 还在 2 [D] → 拉取 D
1000ms A 完成，running-- → _run() 队列空 1 []
1400ms D 完成 0 []

🚀 进阶变体

变体 1：支持任务优先级

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
class PriorityScheduler {
  constructor(max) {
    this.max     = max;
    this.running = 0;
    this.queue   = []; // { promiseCreator, resolve, reject, priority }
  }

  add(promiseCreator, priority = 0) {
    return new Promise((resolve, reject) => {
      this.queue.push({ promiseCreator, resolve, reject, priority });
      // 按优先级降序排列（priority 越大越先执行）
      this.queue.sort((a, b) => b.priority - a.priority);
      this._run();
    });
  }

  _run() {
    while (this.running < this.max && this.queue.length > 0) {
      const { promiseCreator, resolve, reject } = this.queue.shift();
      this.running++;
      promiseCreator()
        .then(resolve, reject)
        .finally(() => { this.running--; this._run(); });
    }
  }
}

变体 2：支持取消（AbortSignal）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class Scheduler {
  constructor(max) {
    this.max     = max;
    this.running = 0;
    this.queue   = [];
  }

  add(promiseCreator, signal) {
    return new Promise((resolve, reject) => {
      if (signal?.aborted) return reject(new DOMException('Aborted', 'AbortError'));

      const task = { promiseCreator, resolve, reject };
      this.queue.push(task);

      signal?.addEventListener('abort', () => {
        const idx = this.queue.indexOf(task);
        if (idx !== -1) {
          this.queue.splice(idx, 1); // 从队列里移除
          reject(new DOMException('Aborted', 'AbortError'));
        }
      }, { once: true });

      this._run();
    });
  }

  _run() {
    while (this.running < this.max && this.queue.length > 0) {
      const { promiseCreator, resolve, reject } = this.queue.shift();
      this.running++;
      promiseCreator()
        .then(resolve, reject)
        .finally(() => { this.running--; this._run(); });
    }
  }
}

变体 3：用 async/await 实现（面试偶尔考）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
class Scheduler {
  constructor(max) {
    this.max   = max;
    this.slots = []; // 存放正在运行的 Promise
  }

  async add(promiseCreator) {
    // 如果槽满了，等待任意一个完成
    if (this.slots.length >= this.max) {
      await Promise.race(this.slots);
    }

    const p = promiseCreator().finally(() => {
      this.slots.splice(this.slots.indexOf(p), 1);
    });
    this.slots.push(p);

    return p;
  }
}

⚠️ 注意：Promise.race 变体在边界情况下有微妙的竞态问题（多个任务同时完成时），生产环境优先用队列版本。

🧠 面试延伸问题

Q：为什么不直接用 Promise.all？
A：Promise.all 是「全部同时发出，全部完成才返回」，没有并发限制。你有 100 个请求会全部同时打出去。

Q：这个模式在实际项目里用在哪？
A：图片/文件批量上传（限制同时上传数）、爬虫（限速防封 IP）、API 限流（遵守第三方 rate limit）。

Q：_run 里为什么用 while 而不是 if？
A：add 时可能槽位不止空出一个（比如批量 resolve）—— 不对，其实每次 finally 只完成一个，while 保证一次调用能把所有空槽填满（比如刚初始化时连续 add 多个任务）。

📌 一句话总结

维护一个「运行计数器」和「等待队列」。每次有任务完成就 -- 然后从队列头取下一个，只要计数器没超过上限就一直取。