实现一个异步并发调度器

January 03, 2021 2430

笔者字节一面的时候遇到了下面这个面试题：

1
2
3
4
5
6
7
//支持并发的调度器， 最多允许2两任务进行处理
const scheduler = new Scheduler(2)
scheduler.addTask(1, '1’);   // 1s后输出’1'
scheduler.addTask(2, '2’);  // 2s后输出’2'
scheduler.addTask(1, '3’);  // 2s后输出’3'
scheduler.addTask(1, '4’);  // 3s后输出’4'
scheduler.start();

当时由于紧张和经验不够，一下子卡壳了，没能及时做出来（痛哭） 回来后好好想了一下，有了以下的思路： 1. 首先用一个maxCount记录允许并发的数量，用一个tasks数组保存所有开始之前设定的任务
2. 我们需要对加入的任务进行一些封装，改成promise，这样就可以获取到完成的状态进行操作 3. 我们用working数组保存正在运行的任务，用tasks保存还未完成的任务 4. 在任务完成后，将tasks队列的第一个任务放入working中，并且运行 5. 如此，通过递归，就可以完成一个并发控制器

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
// 异步调度器
class Scheduler {
    maxCount = 0
    tasks = []
    working = []
    constructor(count) {
        this.maxCount = count;
    }
    addTask = (timer, content) => {
        // 控制函数
        const target = () => {
            return new Promise((resolve) => {
                setTimeout(() => {
                    console.log(content);
                    resolve();
                }, 1000 * timer);
            })
        }
        // 入队列
        this.tasks.push(target)
    }
    continueWork = (fn) => {
        // 递归终点（执行完成)
        if (this.tasks.length > 0) {
            // 将后面的拿进来继续执行
            // 先确定下标
            let idx = -1;
            for (let i = 0; i < this.working.length; i++) {
                if (fn === this.working[i]) {
                    // 将其替换并执行
                    idx = i;
                    break;
                }
            }
            // 调用并运行
            const next = this.tasks.shift();
            next().then(() => {
                this.continueWork(next)
            })
            this.working[idx] = next;
        }
    }
    start = () => {
        let len = this.tasks.length;
        if (len >= this.maxCount) {
            // 否则就将其加入工作队列并执行
            this.working = this.tasks.splice(0, this.maxCount);
            console.log(this.working.length)
            this.working.map(fn => {
                fn().then(() => {
                    // 完成后再回调
                    // 当前执行完毕
                    this.continueWork(fn);
                })
            })
        } else {
            //小于调度范围： 直接全部执行
            this.tasks.map(fn => fn())
        }
    }
}