前言
今天突然想起来之前字节面试碰到的一个前端面试题，题目是实现一个带并发限制的异步调度器Scheduler，这也是网上很多面试字节的前端开发者都碰到过的一个经典题目，今天就来记录一下我的解题思路以及就着这个题目的一些扩展。
题目解答
题目描述
给出的初始代码大致如下，要求实现一个 Scheduler 类，能够控制异步任务的并发执行数量，并符合输出结果要求：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
class Scheduler {
// ...todo
}
const scheduler = new Scheduler(2);
const addTask = (time, name) => {
scheduler.add(time, name);
};
addTask(1000,"1"); // 1000ms后输出1
addTask(500,"2"); // 500ms后输出2
addTask(600,"3"); // 1100ms后输出3
addTask(400,"4"); // 1400ms后输出4
scheduler.start();
|
解题思路
首先，我们需要理解题目的要求：实现一个带并发限制的异步调度器。这意味着我们需要控制同时执行的任务数量，当有任务完成时，再从待执行队列中取出新的任务来执行。所以我们需要维护一个任务队列，并且需要一个变量记录当前正在执行的任务数量，一个变量控制最大并发数。那么我们的基本的变量部分的设计就明确了：
1
2
3
4
5
6
|
class Scheduler {
private limit: number; // 最大并发数
private queue: Array<() => Promise<void>>; // 任务队列
private runningCount: number; // 当前正在执行的任务数量
}
|
随后我们观察题目，可以看到我们需要完善add方法和start方法。add方法用于添加任务到队列中，而start方法用于启动调度器，开始执行任务。start方法应该不需要太复杂，只需要一个循环去检查当前正在执行的任务数量是否小于最大并发数，如果小于，就从队列中取出一个任务执行，并增加当前正在执行的任务数量。当任务完成后，减少当前正在执行的任务数量，并继续检查队列中是否有待执行的任务。大致可以设计如下：
1
2
3
4
5
6
7
8
9
10
11
12
|
start() {
while(this.runningCount < this.limit && this.queue.length > 0) {
this.runningCount++;
const task = this.queue.shift();
// task可能是undefined，所以我们需要加一个可选链操作符来确保安全调用
task?.().finally(() => {
this.runningCount--;
this.start();
});
}
}
|
比较重要的是add方法的设计，根据题目很明显我们需要一个定时器来控制任务的执行时间，并且由于任务得是异步的，所以我们得用一个Promise来包裹任务：
1
2
3
4
5
6
7
8
9
|
add(time:number, name:string) {
this.queue.push(() => new Promise<void>((resolve) => {
setTimeout(() => {
console.log(name);
resolve();
}, time);
}))
}
|
这样我们就完成了Scheduler类的设计，接下来可以展示完整的代码实现。
完整代码
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
|
class Scheduler {
private limit: number; // 最大并发数
private queue: Array<() => Promise<void>>; // 任务队列
private runningCount: number; // 当前正在执行的任务数量
constructor(limit: number) {
this.limit = limit;
this.queue = [];
this.runningCount = 0;
}
add(time: number, name: string) {
this.queue.push(() => new Promise<void>((resolve) => {
setTimeout(() => {
console.log(name);
resolve();
}, time);
}));
}
start() {
while (this.runningCount < this.limit && this.queue.length > 0) {
this.runningCount++;
const task = this.queue.shift();
// task可能是undefined，所以我们需要加一个可选链操作符来确保安全调用
task?.().finally(() => {
this.runningCount--;
this.start();
});
}
}
}
|
扩展一下
优化任务添加
我们发现，使用start方法需要手动调用，并且必须得把所有的任务都添加完毕之后才能调用start方法，这样就不太符合实际的使用场景了。我们可以在add方法中添加任务后自动执行任务，这样就不需要手动调用start方法了。我们可以在add方法中调用start方法来实现这个功能。
1
2
3
4
5
6
7
8
9
|
add(time: number, name: string) {
this.queue.push(() => new Promise<void>((resolve) => {
setTimeout(() => {
console.log(name);
resolve();
}, time);
}));
this.start(); // 添加任务后自动执行任务
}
|
优化add方法
这样就不需要自己.start()了，直接add就可以了。但是这样的缺点还是很明显，首先我们硬编码了setTimeout，这样不管什么任务都会被setTimeout包裹，这样就不太灵活了。其次我们发现这样写我们完全无法获取到任务的返回值，也就是我们完全无法知道任务真正完成的时间，且这样不支持我们在函数外部使用await来等待任务的完成。所以我们其实可以扩展一下题目，让add方法返回一个Promise，并且改写add方法接收的参数，让add方法接收一个函数，这个函数返回一个Promise，这样我们就可以在add方法中直接调用这个函数来获取Promise，并且我们可以在外部使用await来等待任务的完成。结构大致如下：
1
2
3
4
5
6
7
8
9
10
11
|
add(task: () => Promise<void>): Promise<void> {
return new Promise<void>((resolve) => {
this.queue.push(() => {
return task().finally(() => {
resolve();
});
});
this.start();
})
}
|
改造addTask方法
然后改造一下addTask方法，让其可以接收time和name参数：
1
2
3
4
5
6
7
8
9
10
11
|
const addTask = (time, name) => {
return scheduler.add(() => {
return new Promise<void>((resolve) => {
setTimeout(() => {
console.log(name);
resolve();
}, time); // setTimeout本身是非阻塞的，所以需要在外部包裹Promise来实现异步任务的等待
})
})
};
|
完整代码
完整的代码如下：
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
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
|
class Scheduler {
private limit: number; // 最大并发数
private queue: Array<() => Promise<void>>; // 任务队列
private runningCount: number; // 当前正在执行的任务数量
constructor(limit: number) {
this.limit = limit;
this.queue = [];
this.runningCount = 0;
}
add(task: () => Promise<void>): Promise<void> {
return new Promise<void>((resolve) => {
this.queue.push(() => {
return task().finally(() => {
resolve();
});
});
this.start();
})
}
private start() {
while (this.runningCount < this.limit && this.queue.length > 0) {
this.runningCount++;
const task = this.queue.shift();
task?.().finally(() => {
this.runningCount--;
this.start();
});
}
}
}
const scheduler = new Scheduler(2);
const addTask = (time, name) => {
return scheduler.add(() => {
return new Promise<void>((resolve) => {
setTimeout(() => {
console.log(name);
resolve();
}, time); // setTimeout本身是非阻塞的，所以需要在外部包裹Promise来实现异步任务的等待
})
})
};
addTask(1000,"1"); // 1000ms后输出1
addTask(500,"2"); // 500ms后输出2
addTask(600,"3"); // 1100ms后输出3
addTask(400,"4"); // 1400ms后输出4
|
原理解析
并发的过程如下：
- Task 1 入队：queue 存入 Task 1，调用 start()。start() 检查：runningCount (0) < limit (2) 成立，runningCount 变为 1。立刻执行 Task 1 的函数 setTimeout(…, 1000) 被丢给浏览器的定时器线程。
- Task 2 入队：queue 存入 Task 2，调用 start()。start() 检查：runningCount (1) < limit (2) 依然成立，runningCount 变为 2。立刻执行 Task 2 的函数 setTimeout(…, 500) 也被丢给定时器线程。
- Task 3 入队：queue 存入 Task 3，调用 start()。start() 检查：runningCount (2) < limit (2) 不成立（容量已满）。while 循环直接跳过，Task 3 安静地停留在 queue 数组中，函数根本没被调用（定时器还没倒计时）。
- Task 4 入队：同理，runningCount 依旧是 2，Task 4 留在 queue 中。Task 1 和 Task 2 的 setTimeout 在 0ms 时同时在后台开始倒计时，它俩此时是并发的。
当来到500ms时：
- Task 2 (500ms) 率先定时结束！
- Task 1 (1000ms) 还在继续倒计时（还剩 500ms）。
- Task 3, 4 依然在 queue 数组里挂起。
当 Task 2 的 setTimeout 触发时，连锁反应发生：Task 2 执行 console.log(“2”)。Task 2 调用 resolve()，触发自己的 .finally()。.finally() 的回调函数被放入微任务队列，立刻执行：this.runningCount–，正在运行数由 2 降为 1（腾出了一个空位）。再次调用 this.start()，重新进入 start() 的 while 循环：条件检查：runningCount (1) < limit (2) 且 queue.length (2) > 0 成立。runningCount 再次变为 2。queue.shift() 弹出 Task 3 并执行它，Task 3 的 setTimeout(…, 600) 正式启动。
最终版
按照这个思路我们可以更加泛化的实现一个带并发限制的异步调度器Scheduler，可以引入泛型，并且优化一下逻辑，代码如下：
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
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
|
class Scheduler {
private limit: number;
private queue: Array<() => void>;
private runningCount: number;
constructor(limit: number) {
this.limit = limit;
this.queue = [];
this.runningCount = 0;
}
add<T>(task: () => Promise<T>): Promise<T> {
return new Promise((resolve, reject) => {
const runTask = () => {
this.runningCount++;
Promise.resolve() // 使用 Promise.resolve() 来确保 task() 的执行是异步的
.then(() => task())
.then(resolve, reject)
.finally(() => {
this.runningCount--;
// 执行下一个任务
this._next();
});
};
if (this.runningCount < this.limit) {
runTask();
} else {
this.queue.push(runTask);
}
});
}
private _next(): void {
if (this.queue.length > 0 && this.runningCount < this.limit) {
const nextTask = this.queue.shift(); // 从队列的开头取出下一个任务
nextTask?.();
}
}
}
// 辅助函数：延迟指定毫秒数并返回结果，用于模拟异步任务
const timeout = (time: number) =>
new Promise((resolve) => setTimeout(resolve, time));
// 实例化一个并发限制为 2 的调度器
const scheduler = new Scheduler(2);
// 定义添加任务的辅助工具函数（打印带时间的日志）
const addTask = (time: number, order: string) => {
const startTime = Date.now();
console.log(`[0ms] 提交任务: ${order}`);
scheduler
.add(async () => {
console.log(`[${Date.now() - startTime}ms] 开始执行: ${order}`);
await timeout(time);
return `任务 ${order} 完成`;
})
.then((res) => {
console.log(`[${Date.now() - startTime}ms] ${res}`);
});
};
// 按照顺序连续提交 4 个任务
addTask(1000, "1"); // 耗时 1000ms
addTask(500, "2"); // 耗时 500ms
addTask(300, "3"); // 耗时 300ms
addTask(400, "4"); // 耗时 400ms
|
结语
上面介绍的就是对带并发限制的异步调度器Scheduler的实现思路和代码实现，整个实现过程主要是通过维护一个任务队列和当前正在执行的任务数量来控制并发执行的数量。希望对大家在面试中遇到类似问题时有所帮助。