【vue3源码】六、scheduler
scheduler即调度器是vue3中比较重要的一个概念。通过scheduler进行调度任务（job），保证了vue中相关API及生命周期函数、组件渲染顺序的正确性。
我们知道，使用watchEffect进行侦听数据源时，侦听器将会在组件渲染之前执行；watchSyncEffect进行侦听数据源时，侦听器在依赖发生变化后立即执行；而watchPostEffect进行侦听数据源时，侦听器会在组件渲染后才执行。针对不同的侦听器的执行顺序，就是通过scheduler进行统一调度而实现的。
scheduler的实现
在scheduler中主要通过三个队列实现任务调度，这三个对列分别为：
-
pendingPreFlushCbs：组件更新前置任务队列 -
queue：组件更新任务队列 -
pendingPostFlushCbs：组件更新后置任务队列
如何使用这几个队列？vue中有三个方法分别用来进行对pendingPreFlushCbs、queue、pendingPostFlushCbs入队操作。
// 前置任务队列入队
export function queuePreFlushCb(cb: SchedulerJob) {
queueCb(cb, activePreFlushCbs, pendingPreFlushCbs, preFlushIndex)
}
// 后置任务队列入队
export function queuePostFlushCb(cb: SchedulerJobs) {
queueCb(cb, activePostFlushCbs, pendingPostFlushCbs, postFlushIndex)
}
function queueCb(
cb: SchedulerJobs,
activeQueue: SchedulerJob[] | null,
pendingQueue: SchedulerJob[],
index: number
) {
// 如果cb不是数组
if (!isArray(cb)) {
// 激活队列为空或cb不在激活队列中，需要将cb添加到对应队列中
if (
!activeQueue ||
!activeQueue.includes(cb, cb.allowRecurse ? index + 1 : index)
) {
pendingQueue.push(cb)
}
}
// cb是数组
else {
// 如果 cb 是一个数组，那么它是一个组件生命周期钩子
// 其已经被去重了，因此我们可以在此处跳过重复检查以提高性能
pendingQueue.push(...cb)
}
queueFlush()
}
// queue队列入队
export function queueJob(job: SchedulerJob) {
// 当满足以下情况中的一种才可以入队
// 1. queue长度为0
// 2. queue中不存在job（如果job是watch()回调，搜索从flushIndex + 1开始，否则从flushIndex开始），并且job不等于currentPreFlushParentJob
if (
(!queue.length ||
!queue.includes(
job,
isFlushing && job.allowRecurse ? flushIndex + 1 : flushIndex
)) &&
job !== currentPreFlushParentJob
) {
// job.id为null直接入队
if (job.id == null) {
queue.push(job)
} else {
// 插队，插队后queue索引区间[flushIndex + 1, end]内的job.id是非递减的
// findInsertionIndex方法通过二分法寻找[flushIndex + 1, end]区间内大于等于job.id的第一个索引
queue.splice(findInsertionIndex(job.id), 0, job)
}
queueFlush()
}
}
三个队列的入队几乎是相似的，都不允许队列中存在重复job（从队列的flushIndex或flushIndex + 1开始搜索）。不同的是queue允许插队。
queueFlush
job入队后，会调用一个queueFlush函数：
function queueFlush() {
// isFlushing表示是否正在执行队列
// isFlushPending表示是否正在等待执行队列
// 如果此时未在执行队列也没有正在等待执行队列，则需要将isFlushPending设置为true，表示队列进入等待执行状态
// 同时在下一个微任务队列执行flushJobs，即在下一个微任务队列执行队列
if (!isFlushing && !isFlushPending) {
isFlushPending = true
currentFlushPromise = resolvedPromise.then(flushJobs)
}
}
为什么需要将flushJobs放入下一个微任务队列，而不是宏任务队列？
首先微任务比宏任务有更高的优先级，当同时存在宏任务和微任务时，会先执行全部的微任务，然后再执行宏任务，这说明通过微任务，可以将flushJobs尽可能的提前执行。如果使用宏任务，如果在queueJob之前有多个宏任务，则必须等待这些宏任务执行完后，才能执行queueJob，这样以来flushJobs的执行就会非常靠后。
flushJobs
在flushJobs中会依次执行pendingPreFlushCbs、queue、pendingPostFlushCbs中的任务，如果此时还有剩余job，则继续执行flushJobs，知道将三个队列中的任务都执行完。
function flushJobs(seen?: CountMap) {
// 将isFlushPending置为false，isFlushing置为true
// 因为此时已经要开始执行队列了
isFlushPending = false
isFlushing = true
if (__DEV__) {
seen = seen || new Map()
}
// 执行前置任务队列
flushPreFlushCbs(seen)
// queue按job.id升序排列
// 这可确保：
// 1. 组件从父组件先更新然后子组件更新。（因为 parent 总是在 child 之前创建，所以它的redner effect会具有较高的优先级）
// 2. 如果在 parent 组件更新期间卸载组件，则可以跳过其更新
queue.sort((a, b) => getId(a) - getId(b))
// 用于检测是否是无限递归，最多 100 层递归，否则就报错，只会开发模式下检查
const check = __DEV__
? (job: SchedulerJob) => checkRecursiveUpdates(seen!, job)
: NOOP
try {
// 执行queue中的任务
for (flushIndex = 0; flushIndex < queue.length; flushIndex++) {
const job = queue[flushIndex]
if (job && job.active !== false) {
if (__DEV__ && check(job)) {
continue
}
callWithErrorHandling(job, null, ErrorCodes.SCHEDULER)
}
}
} finally {
// 清空queue并将flushIndex重置为0
flushIndex = 0
queue.length = 0
// 执行后置任务队列
flushPostFlushCbs(seen)
// 将isFlushing置为false，说明此时任务已经执行完
isFlushing = false
currentFlushPromise = null
// 执行剩余job
// post队列执行过程中可能有job加入，继续调用flushJobs执行剩余job
if (
queue.length ||
pendingPreFlushCbs.length ||
pendingPostFlushCbs.length
) {
flushJobs(seen)
}
}
}
flushPreFlushCbs
flushPreFlushCbs用来执行pendingPreFlushCbs中的job。
export function flushPreFlushCbs(
seen?: CountMap,
parentJob: SchedulerJob | null = null
) {
// 有job才执行
if (pendingPreFlushCbs.length) {
// 赋值父job
currentPreFlushParentJob = parentJob
// 去重并将队列赋值给activePreFlushCbs
activePreFlushCbs = [...new Set(pendingPreFlushCbs)]
// 清空pendingPreFlushCbs
pendingPreFlushCbs.length = 0
if (__DEV__) {
seen = seen || new Map()
}
// 循环执行job
for (
preFlushIndex = 0;
preFlushIndex < activePreFlushCbs.length;
preFlushIndex++
) {
if (
__DEV__ &&
checkRecursiveUpdates(seen!, activePreFlushCbs[preFlushIndex])
) {
continue
}
activePreFlushCbs[preFlushIndex]()
}
// 执行完毕后将activePreFlushCbs重置为null、preFlushIndex重置为0、currentPreFlushParentJob重置为null
activePreFlushCbs = null
preFlushIndex = 0
currentPreFlushParentJob = null
// 递归flushPreFlushCbs，直到pendingPreFlushCbs为空停止
flushPreFlushCbs(seen, parentJob)
}
}
flushPostFlushCbs
export function flushPostFlushCbs(seen?: CountMap) {
// flush any pre cbs queued during the flush (e.g. pre watchers)
flushPreFlushCbs()
// 存在job才执行
if (pendingPostFlushCbs.length) {
// 去重
const deduped = [...new Set(pendingPostFlushCbs)]
// 清空pendingPostFlushCbs
pendingPostFlushCbs.length = 0
// #1947 already has active queue, nested flushPostFlushCbs call
// 已经存在activePostFlushCbs，嵌套flushPostFlushCbs调用，直接return
if (activePostFlushCbs) {
activePostFlushCbs.push(...deduped)
return
}
activePostFlushCbs = deduped
if (__DEV__) {
seen = seen || new Map()
}
// 按job.id升序
activePostFlushCbs.sort((a, b) => getId(a) - getId(b))
// 循环执行job
for (
postFlushIndex = 0;
postFlushIndex < activePostFlushCbs.length;
postFlushIndex++
) {
if (
__DEV__ &&
checkRecursiveUpdates(seen!, activePostFlushCbs[postFlushIndex])
) {
continue
}
activePostFlushCbs[postFlushIndex]()
}
// 重置activePostFlushCbs及、postFlushIndex
activePostFlushCbs = null
postFlushIndex = 0
}
}
nextTick
export function nextTick<T = void>(
this: T,
fn?: (this: T) => void
): Promise<void> {
const p = currentFlushPromise || resolvedPromise
return fn ? p.then(this ? fn.bind(this) : fn) : p
}
nextTick会在flushJobs执行完成后才会执行，组件的更新及onUpdated、onMounted等某些生命周期钩子会在nextTick之前执行。所以在nextTick.then中可以获取到最新的DOM。
哪些操作会交给调度器进行调度？
-
watchEffect、watchPostEffect，分别会将侦听器的执行加入到前置任务队列与后置任务队列。
function doWatch() {
// ...
const job: SchedulerJob = () => {
if (!effect.active) {
return
}
if (cb) {
// watch(source, cb)
const newValue = effect.run()
if (
deep ||
forceTrigger ||
(isMultiSource
? (newValue as any[]).some((v, i) =>
hasChanged(v, (oldValue as any[])[i])
)
: hasChanged(newValue, oldValue)) ||
(__COMPAT__ &&
isArray(newValue) &&
isCompatEnabled(DeprecationTypes.WATCH_ARRAY, instance))
) {
// cleanup before running cb again
if (cleanup) {
cleanup()
}
callWithAsyncErrorHandling(cb, instance, ErrorCodes.WATCH_CALLBACK, [
newValue,
// pass undefined as the old value when it's changed for the first time
oldValue === INITIAL_WATCHER_VALUE ? undefined : oldValue,
onCleanup
])
oldValue = newValue
}
} else {
// watchEffect
effect.run()
}
}
if (flush === 'sync') {
scheduler = job as any // the scheduler function gets called directly
} else if (flush === 'post') {
scheduler = () => queuePostRenderEffect(job, instance && instance.suspense)
} else {
// default: 'pre'
scheduler = () => queuePreFlushCb(job)
}
const effect = new ReactiveEffect(getter, scheduler)
// ...
}
- 组件的更新函数：
const setupRenderEffect = () => {
// ...
const componentUpdateFn = () => {
//...
}
const effect = (instance.effect = new ReactiveEffect(
componentUpdateFn,
() => queueJob(update),
instance.scope
))
const update: SchedulerJob = (instance.update = () => effect.run())
update.id = instance.uid
// ...
}
-
onMounted、onUpdated、onUnmounted、Transition的enter钩子等一些钩子函数会被放到后置任务队列
export const queuePostRenderEffect = __FEATURE_SUSPENSE__
? queueEffectWithSuspense
: queuePostFlushCb
const mountElement = () => {
// ...
if (
(vnodeHook = props && props.onVnodeMounted) ||
needCallTransitionHooks ||
dirs
) {
queuePostRenderEffect(() => {
vnodeHook && invokeVNodeHook(vnodeHook, parentComponent, vnode)
needCallTransitionHooks && transition!.enter(el)
dirs && invokeDirectiveHook(vnode, null, parentComponent, 'mounted')
}, parentSuspense)
}
}
const patchElement = () => {
// ...
if ((vnodeHook = newProps.onVnodeUpdated) || dirs) {
queuePostRenderEffect(() => {
vnodeHook && invokeVNodeHook(vnodeHook, parentComponent, n2, n1)
dirs && invokeDirectiveHook(n2, n1, parentComponent, 'updated')
}, parentSuspense)
}
}
总结
scheduler通过三个队列实现，在vue只需通过调用queuePreFlushCb、queuePostFlushCb、queueJob方法将job添加到对应队列中，不需要手动控制job的执行时机，完全将job的执行时机交给了scheduler进行调度。
三个队列的特点：
pendingPreFlushCbs |
queue |
pendingPostFlushCbs |
|
|---|---|---|---|
| 执行时机 | DOM更新前 |
queue中的job就包含组件的更新 |
DOM更新后 |
| 是否允许插队 | 不允许 | 允许 | 不允许 |
job执行顺序 |
按入队顺序执行，先进先出 | 按job.id升序顺序执行job。保证父子组件的更新顺序 |
按job.id升序顺序执行job
|
scheduler中通过Promise.resolve()将队列中job的执行（即flushJobs）放入到下一个微任务队列中，而nextTick.then中回调的执行又会被放到下一个微任务队列。等到nextTick.then中回调的执行，队列中的job已经执行完毕，此时DOM已经更新完毕，所以在nextTick.then中就可以获取到更新后的DOM。
