scheduler调度器

# 概览

在vue3的runtime-core模块中，调度器scheduler负责异步更新队列，包括组件的更新和副作用的刷新执行等。其实现文件路径为：packages\runtime-core\src\scheduler.ts

调度器scheduler负责调度的是任务job，所谓的job就是会执行副作用，比如在watchEffect(cb)中,job本质上就是effect.run()即副作用的执行，而副作用effect就是ReactiveEffect(cb)的实例对象，而watch中的job除了执行副作用外，还会执行回调。

在watch和watchEffect，会将job放入队列中，二者在某些情况会有所区别；默认情况下，watch和watchEffect都是通过queueJob方法将job放入队列中，但是watchPostEffect则是通过queuePostRenderEffect。本文会解析调度器scheduler的具体实现，就会理解二者处理job的区别。

# 源码解析

# 变量

const queue = []; // 存储待执行的更新任务
const pendingPostFlushCbs = []; // 存储待执行的后置刷新回调，通常是在DOM更新后执行
let activePostFlushCbs = null; // 当前正在执行的后置刷新回调
let flushIndex = -1; // 当前正在刷新的任务在queue中的索引
let postFlushIndex = 0; // 当前正在执行的后置刷新回调在activePostFlushCbs中的索引
let currentFlushPromise = null; // 当前刷新队列的Promise。用于确保刷新操作是异步的，并且同一时间只有一个刷新操作在执行。
const resolvePromise = Promise.resolve(); // 一个已经resolved的Promise，用于创建微任务

1
2
3
4
5
6
7

# 核心方法

# `queueJob`

queueJob方法用于将任务Job加入到队列。

function queueJob(job) {
    // 首先判断job是否加入过队列，若没加入队列，就继续执行后续
    if (!(job.flags & 1)) {
        // 获取job的id，该id和组件实例id等同
        const jobId = getId(job);
        // 获取队列中的最后一个任务
        const lastJob = queue[queue.length - 1];
        
        if (!lastJob || !(job.flags & 2) && jobId >= getId(lastJob)) {
            // 若lastJob不存在，即queue为空，或者job的标志flags表示是一个前置任务并且jobId不小于lastJob的id，则直接将job插入到queue的末尾
            queue.push(job);
        } else {
            // 否则调用findInsertionIndex，根据任务的id找到合适的插入位置
            queue.splice(findInsertionIndex(jobId), 0, job)
        }

        job.flags |= 1; // 修改任务的flags,将其标记为已插入queue队列
        queueFlush(); // 最后调用queueFlush刷新队列
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# `queueFlush`

queueFlush用于创建微任务刷新队列，通过currentFlushPromise确保同一事件循环只创建一次微任务。

function queueFlush() {
    // 判断当前是否有刷新操作，若没有，则创建一个微任务
    if (!currentFlushPromise) {
        currentFlushPromise = resolvePromise.then(flushJobs);
    }
}

1
2
3
4
5
6

queueFlush的调用时机是当有新的任务加入队列(通过queueJob)或新的后置回调queuePostFlushCb时，都会被调用。

# `flushJobs`

flushJobs的作用就是刷新队列，执行所有任务队列。

function flushJobs() {
    try {
        // 遍历任务队列queue
        for (flushIndex = 0; flushIndex < queue.length; flushIndex++) {
            const job = queue[flushIndex];
            // 若任务存在且没有标记为跳过，则执行
            if (job && !(job.flags & 8)) {
                // 若任务是允许递归的，则清除其任务标记，方便后续被重新加入队列
                if (job.flags & 4) {
                    job.flags &= ~1;
                }
                // 通过callWithErrorHandling执行job
                callWithErrorHandling(job, job.i, job.i ? 15 : 14);
                // 执行完任务后，清除其任务标记
                if (!(job.flags & 4)) {
                    job.flags &= ~1;
                }
            }
        }
    } finally {
        // 再次遍历任务队列，清理队列的残留任务
        for (; flushIndex < queue.length; flushIndex++) {
            const job = queue[flushIndex];
            if (job) {
                job.flags &= -2;
            }
        }
        // 重置索引和任务队列
        flushIndex = -1;
        queue.length = 0;
        // 执行后置回调
        flushPostFlushCbs();
        // 将微任务标记置为null
        currentFlushPromise = null;
        // 检查queue队列，若队列不为空，则调用flushJobs再次触发刷新
        if (queue.length || pendingPostFlushCbs.length) {
            flushJobs();
        }
    }
}

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

flushJobs就是遍历任务队列，将任务通过callWithErrorHandling执行。

# `queuePostFlushCb`

watchPostEffect中是通过queuePostRenderEffect将任务放入后置任务队列中，而queuePostRenderEffect只是queueEffectWithSuspense方法的别名。

queueEffectWithSuspense可以处理异步组件的副作用，其实现如下：

function queueEffectWithSuspense(fn, suspense) {
   // 判断suspense(组件实例)和实例是否处于pending状态
    if (suspense && suspense.pendingBranch) {
        // 若是异步组件,且处于pending状态，则判断job是否是数组；将任务job加入到异步组件的effects中
        if (shared.isArray(fn)) {
            suspense.effects.push(...fn)
        } else {
            suspense.effects.push(fn)
        }
    } else {
      // 若不是异步组件或者异步组件不处于pending状态，则调用queuePostFlushCb
        queuePostFlushCb(fn)
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14

queuePostFlushCB方法就是将回调加入到后置队列中。

function queuePostFlushCb(cb) {
    // 判断回调是否是数组
    if (!shared.isArray(cb)) {
        // 若是单个回调，则判断是否正在执行后置回调，若正在执行，且回调是一个前置任务 
        if (activePostFlushCbs && cb.id === -1) {
            // 则将回调任务插入到当前执行位置之后
            activePostFlushCbs.splice(postFlushIndex + 1, 0, cb)
        } else if (!(cb.flags & 1)) {
            // 判断任务的标记是否被添加过，若没添加，则将其添加，并且修改标记
            pendingPostFlushCbs.push(cb);
            cb.flags |= 1;
        }
    } else {
        // 若是数组，将其添加到后置队列pendingPostFlushCbs中 
        pendingPostFlushCbs.push(...cb)
    }
    // 最后调用queueFlush创建微任务队列
    queueFlush()
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

# `flushPostFlushCbs`

flushPostFlushCbs用于刷新后置回调，执行回调。其实现如下：

function flushPostFlushCbs() { 
    // 判断后置任务队列是否为空
    if (pendingPostFlushCbs.length) {
        // 使用Set去重后置任务队列，并且根据任务的id升序排序
        const deduped = [...new Set(pendingPostFlushCbs)].sort((a, b) => getId(a) - getId(b))
        // 清空后置任务队列
        pendingPostFlushCbs.length = 0;
        
        // 若当前存在激活的后置队列，则将去重排序后的回调数组追加到当前激活的回调中，然后返回，避免重复执行。
        if (activePostFlushCbs) {
            activePostFlushCbs.push(...deduped);
            return;
        }
        // 若当前不存在激活的后置队列，将去重排序后的回调数组赋值给activePostFlushCbs
        activePostFlushCbs = deduped;
        // 遍历激活的后置队列，检查标记位,执行回调
        for (postFlushIndex = 0; postFlushIndex < activePostFlushCbs.length; postFlushIndex++) {
            const cb = activePostFlushCbs[postFlushIndex];
            if (cb.flags & 4) {
                cb.flags &= -2;
            }
            if (!(cb.flags & 8)) {
                cb();
            }
            cb.flags &= -2;
        }
        // 最后将activePostFlushCbs置为null，索引置为0
        activePostFlushCbs = null;
        postFlushIndex = 0;
    }
}

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

flushPostFlushCbs的调用时机就是一般在flushJobs执行完后执行，或执行完前置任务队列后。

# `flushPreFlushCbs`

flushPreFlushCbs用于刷新前置回调，比如默认情况下的watch(source,cb)和watchEffect(effect)

flushPreFlushCbs的实现如下：

function flushPreFlushCbs(instance, seen, i = flushIndex + 1) {
    // 遍历任务队列，跳过当前的任务
    for (; i < queue.length; i++) {
        const cb = queue[i];
        // 判断cb回调存在，并且回调是一个前置任务
        if (cb && cb.flags & 2) {
            // 若实例存在，并且实例的id和回调的id不一样，则跳过执行
            if (instance && cb.id !== instance.id) {
                continue;
            }
            // 从队列中移除回调任务
            queue.splice(i, 1);
            // 索引自减
            i--;
            // 若回调的标志位允许递归，则清除标志位
            if (cb.flags & 4) {
                cb.flags &= -2;
            }
            // 执行回调
            cb();
            // 若回调的标志位不允许递归，则清除标志位
            if (!(cb.flags & 4)) {
                cb.flags &= -2;
            }
        }
    }
}

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

flushPreFlushCbs只执行该实例的回调任务，在组件更新前会被调用。

# 辅助方法

调度器中的辅助方法包含getId和findInsertionIndex；

getId

getId用于获取任务的id，若id存在，则返回id；否则，判断任务的flags，即任务是否是一个前值任务，若是前置任务，则返回*-1*，表示其优先级很高，若不是前置任务，则返回一个正无穷大。

const getId = (job) => job.id == null ? job.flags & 2 ? -1 : Infinity : job.id;

findInsertionIndex

findInsertionIndex就是通过二分法根据id在queue队列中找到一个合适的索引位置。

function findInsertionIndex(id) {
    let start = flushIndex + 1;
    let end = queue.length;
    while (start < end) {
        const middle = start + end >>> 1;
        const middleJob = queue[middle];
        // 返回中间索引的任务
        const middleJobId = getId(middleJob);
        if (middleJobId < id || middleJobId === id && middleJob.flags & 2) {
            // 相同id，即同一个实例的任务，前置任务在前面
            start = middle + 1;
        } else {
            end = middle;
        }
    }
    return start;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

编辑

上次更新: 2025/09/12, 09:24:28

← watch和watcherEffect源码解析 directive自定义指令实现原理→

scheduler调度器

# 概览

# 源码解析

# 变量

# 核心方法

# queueJob

# queueFlush

# flushJobs

# queuePostFlushCb

# flushPostFlushCbs

# flushPreFlushCbs