09 | 侦听器：侦听器的实现原理和使用场景是什么？（下）

Vue.js 3.0 核心源码解析 - 前百度、滴滴资深技术专家 - 拉勾教育

在前面的课时中，我们多次提到回调函数是以一种调度的方式执行的，特别是当 flush 不是 sync 时，它会把回调函数执行的任务推到一个异步队列中执行。接下来，我们就来分析异步执行队列的设计。分析之前，我们先来思考一下，为什么会需要异步队列？

异步任务队列的设计

我们把之前的例子简单修改一下：

import { reactive, watch } from 'vue' 
const state = reactive({ count: 0 }) 
watch(() => state.count, (count, prevCount) => { 
  console.log(count) 
}) 
state.count++ 
state.count++ 
state.count++

这里，我们修改了三次 state.count，那么 watcher 的回调函数会执行三次吗？

答案是不会，实际上只输出了一次 count 的值，也就是最终计算的值 3。这在大多数场景下都是符合预期的，因为在一个 Tick（宏任务执行的生命周期）内，即使多次修改侦听的值，它的回调函数也只执行一次。

知识延伸
组件的更新过程是异步的，我们知道修改模板中引用的响应式对象的值时，会触发组件的重新渲染，但是在一个 Tick 内，即使你多次修改多个响应式对象的值，组件的重新渲染也只执行一次。这是因为如果每次更新数据都触发组件重新渲染，那么重新渲染的次数和代价都太高了。

那么，这是怎么做到的呢？我们先从异步任务队列的创建说起。

异步任务队列的创建

通过前面的分析我们知道，在创建一个 watcher 时，如果配置 flush 为 pre 或不配置 flush ，那么 watcher 的回调函数就会异步执行。此时分别是通过 queueJob 和 queuePostRenderEffect 把回调函数推入异步队列中的。

在不涉及 suspense 的情况下，queuePostRenderEffect 相当于 queuePostFlushCb，我们来看它们的实现：

const queue = [] 
const postFlushCbs = [] 
function queueJob(job) { 
if (!queue.includes(job)) { 
    queue.push(job) 
    queueFlush() 
  } 
} 
function queuePostFlushCb(cb) { 
if (!isArray(cb)) { 
    postFlushCbs.push(cb) 
  } 
else { 
    postFlushCbs.push(...cb) 
  } 
  queueFlush() 
}

Vue.js 内部维护了一个 queue 数组和一个 postFlushCbs 数组，其中 queue 数组用作异步任务队列， postFlushCbs 数组用作异步任务队列执行完毕后的回调函数队列。

执行 queueJob 时会把这个任务 job 添加到 queue 的队尾，而执行 queuePostFlushCb 时，会把这个 cb 回调函数添加到 postFlushCbs 的队尾。它们在添加完毕后都执行了 queueFlush 函数，我们接着看它的实现：

const p = Promise.resolve() 
let isFlushing = false 
let isFlushPending = false 
function nextTick(fn) { 
return fn ? p.then(fn) : p 
} 
function queueFlush() { 
if (!isFlushing && !isFlushPending) { 
    isFlushPending = true 
nextTick(flushJobs) 
  } 
}

可以看到，Vue.js 内部还维护了 isFlushing 和 isFlushPending 变量，用来控制异步任务的刷新逻辑。

在 queueFlush 首次执行时，isFlushing 和 isFlushPending 都是 false，此时会把 isFlushPending 设置为 true，并且调用 nextTick(flushJobs) 去执行队列里的任务。

因为 isFlushPending 的控制，这使得即使多次执行 queueFlush，也不会多次去执行 flushJobs。另外 nextTick 在 Vue.js 3.0 中的实现也是非常简单，通过 Promise.resolve().then 去异步执行 flushJobs。

因为 JavaScript 是单线程执行的，这样的异步设计使你在一个 Tick 内，可以多次执行 queueJob 或者 queuePostFlushCb 去添加任务，也可以保证在宏任务执行完毕后的微任务阶段执行一次 flushJobs。

异步任务队列的执行

创建完任务队列后，接下来要异步执行这个队列，我们来看一下 flushJobs 的实现：

const getId = (job) => (job.id == null ? Infinity : job.id) 
function flushJobs(seen) { 
  isFlushPending = false 
  isFlushing = true 
  let job 
if ((process.env.NODE_ENV !== 'production')) { 
    seen = seen || new Map() 
  } 
  queue.sort((a, b) => getId(a) - getId(b)) 
while ((job = queue.shift()) !== undefined) { 
if (job === null) { 
continue 
    } 
if ((process.env.NODE_ENV !== 'production')) { 
      checkRecursiveUpdates(seen, job) 
    } 
    callWithErrorHandling(job, null, 14 ) 
  } 
  flushPostFlushCbs(seen) 
  isFlushing = false 
if (queue.length || postFlushCbs.length) { 
    flushJobs(seen) 
  } 
}

可以看到，flushJobs 函数开始执行的时候，会把 isFlushPending 重置为 false，把 isFlushing 设置为 true 来表示正在执行异步任务队列。

对于异步任务队列 queue，在遍历执行它们前会先对它们做一次从小到大的排序，这是因为两个主要原因：

• 我们创建组件的过程是由父到子，所以创建组件副作用渲染函数也是先父后子，父组件的副作用渲染函数的 effect id 是小于子组件的，每次更新组件也是通过 queueJob 把 effect 推入异步任务队列 queue 中的。所以为了保证先更新父组再更新子组件，要对 queue 做从小到大的排序。
• 如果一个组件在父组件更新过程中被卸载，它自身的更新应该被跳过。所以也应该要保证先更新父组件再更新子组件，要对 queue 做从小到大的排序。

接下来，就是遍历这个 queue，依次执行队列中的任务了，在遍历过程中，注意有一个 checkRecursiveUpdates 的逻辑，它是用来在非生产环境下检测是否有循环更新的，它的作用我们稍后会提。

遍历完 queue 后，又会进一步执行 flushPostFlushCbs 方法去遍历执行所有推入到 postFlushCbs 的回调函数：

function flushPostFlushCbs(seen) { 
if (postFlushCbs.length) { 
const cbs = [...new Set(postFlushCbs)] 
    postFlushCbs.length = 0 
if ((process.env.NODE_ENV !== 'production')) { 
      seen = seen || new Map() 
    } 
for (let i = 0; i < cbs.length; i++) { 
if ((process.env.NODE_ENV !== 'production')) {                                                       
        checkRecursiveUpdates(seen, cbs[i]) 
      } 
      cbs[i]() 
    } 
  } 
}

注意这里遍历前会通过 const cbs = [...new Set(postFlushCbs)] 拷贝一个 postFlushCbs 的副本，这是因为在遍历的过程中，可能某些回调函数的执行会再次修改 postFlushCbs，所以拷贝一个副本循环遍历则不会受到 postFlushCbs 修改的影响。

遍历完 postFlushCbs 后，会重置 isFlushing 为 false，因为一些 postFlushCb 执行过程中可能会再次添加异步任务，所以需要继续判断如果 queue 或者 postFlushCbs 队列中还存在任务，则递归执行 flushJobs 把它们都执行完毕。

检测循环更新

前面我们提到了，在遍历执行异步任务和回调函数的过程中，都会在非生产环境下执行 checkRecursiveUpdates 检测是否有循环更新，它是用来解决什么问题的呢？

我们把之前的例子改写一下：

import { reactive, watch } from 'vue' 
const state = reactive({ count: 0 }) 
watch(() => state.count, (count, prevCount) => { 
  state.count++ 
  console.log(count) 
}) 
state.count++

如果你去跑这个示例，你会在控制台看到输出了 101 次值，然后报了错误： Maximum recursive updates exceeded 。这是因为我们在 watcher 的回调函数里更新了数据，这样会再一次进入回调函数，如果我们不加任何控制，那么回调函数会一直执行，直到把内存耗尽造成浏览器假死。
为了避免这种情况，Vue.js 实现了 checkRecursiveUpdates 方法：

const RECURSION_LIMIT = 100 
function checkRecursiveUpdates(seen, fn) { 
if (!seen.has(fn)) { 
    seen.set(fn, 1) 
  } 
else { 
const count = seen.get(fn) 
if (count > RECURSION_LIMIT) { 
throw new Error('Maximum recursive updates exceeded. ' + 
"You may have code that is mutating state in your component's " + 
'render function or updated hook or watcher source function.') 
    } 
else { 
      seen.set(fn, count + 1) 
    } 
  } 
}

通过前面的代码，我们知道 flushJobs 一开始便创建了 seen，它是一个 Map 对象，然后在 checkRecursiveUpdates 的时候会把任务添加到 seen 中，记录引用计数 count，初始值为 1，如果 postFlushCbs 再次添加了相同的任务，则引用计数 count 加 1，如果 count 大于我们定义的限制 100 ，则说明一直在添加这个相同的任务并超过了 100 次。那么，Vue.js 会抛出这个错误，因为在正常的使用中，不应该出现这种情况，而我们上述的错误示例就会触发这种报错逻辑。

优化：只用一个变量

到这里，异步队列的设计就介绍完毕了，你可能会对 isFlushPending 和 isFlushing 有些疑问，为什么需要两个变量来控制呢？

从语义上来看，isFlushPending 用于判断是否在等待 nextTick 执行 flushJobs，而 isFlushing 是判断是否正在执行任务队列。

从功能上来看，它们的作用是为了确保以下两点：

1. 在一个 Tick 内可以多次添加任务到队列中，但是任务队列会在 nextTick 后执行；
2. 在执行任务队列的过程中，也可以添加新的任务到队列中，并且在当前 Tick 去执行剩余的任务队列。

但实际上，这里我们可以进行优化。在我看来，这里用一个变量就足够了，我们来稍微修改一下源码：

function queueFlush() { 
if (!isFlushing) { 
    isFlushing = true 
nextTick(flushJobs) 
  } 
} 
function flushJobs(seen) { 
let job 
if ((process.env.NODE_ENV !== 'production')) { 
    seen = seen || new Map() 
  } 
  queue.sort((a, b) => getId(a) - getId(b)) 
while ((job = queue.shift()) !== undefined) { 
if (job === null) { 
continue 
    } 
if ((process.env.NODE_ENV !== 'production')) { 
      checkRecursiveUpdates(seen, job) 
    } 
    callWithErrorHandling(job, null, 14 ) 
  } 
  flushPostFlushCbs(seen) 
if (queue.length || postFlushCbs.length) { 
    flushJobs(seen) 
  } 
  isFlushing = false 
}

可以看到，我们只需要一个 isFlushing 来控制就可以实现相同的功能了。在执行 queueFlush 的时候，判断 isFlushing 为 false，则把它设置为 true，然后 nextTick 会执行 flushJobs。在 flushJobs 函数执行完成的最后，也就是所有的任务（包括后添加的）都执行完毕，再设置 isFlushing 为 false。

我这么修改源码后也跑通了 Vue.js 3.0 的单元测试，如果你觉得这么实现有问题的话，欢迎在留言区评论与我讨论。

了解完 watch API 和异步任务队列的设计后，我们再来学习侦听器提供的另一个 API—— watchEffect API。

watchEffect API

watchEffect API 的作用是注册一个副作用函数，副作用函数内部可以访问到响应式对象，当内部响应式对象变化后再立即执行这个函数。

可以先来看一个示例：

import { ref, watchEffect } from 'vue' 
const count = ref(0) 
watchEffect(() => console.log(count.value)) 
count.value++

它的结果是依次输出 0 和 1。
watchEffect 和前面的 watch API 有哪些不同呢？主要有三点：

1. 侦听的源不同 。watch API 可以侦听一个或多个响应式对象，也可以侦听一个 getter 函数，而 watchEffect API 侦听的是一个普通函数，只要内部访问了响应式对象即可，这个函数并不需要返回响应式对象。
2. 没有回调函数 。watchEffect API 没有回调函数，副作用函数的内部响应式对象发生变化后，会再次执行这个副作用函数。
3. 立即执行 。watchEffect API 在创建好 watcher 后，会立刻执行它的副作用函数，而 watch API 需要配置 immediate 为 true，才会立即执行回调函数。

对 watchEffect API 有大体了解后，我们来看一下在我整理的 watchEffect 场景下， doWatch 函数的简化版实现：

function watchEffect(effect, options) { 
return doWatch(effect, null, options); 
} 
function doWatch(source, cb, { immediate, deep, flush, onTrack, onTrigger } = EMPTY_OBJ) { 
  instance = currentInstance; 
  let getter; 
if (isFunction(source)) { 
    getter = () => { 
if (instance && instance.isUnmounted) { 
return; 
      } 
if (cleanup) { 
        cleanup(); 
      } 
return callWithErrorHandling(source, instance, 3 , [onInvalidate]); 
    }; 
  } 
  let cleanup; 
const onInvalidate = (fn) => { 
    cleanup = runner.options.onStop = () => { 
      callWithErrorHandling(fn, instance, 4 ); 
    }; 
  }; 
  let scheduler; 
if (flush === 'sync') { 
    scheduler = invoke; 
  } 
else if (flush === 'pre') { 
    scheduler = job => { 
if (!instance || instance.isMounted) { 
        queueJob(job); 
      } 
else { 
        job(); 
      } 
    }; 
  } 
else { 
    scheduler = job => queuePostRenderEffect(job, instance && instance.suspense); 
  } 
const runner = effect(getter, { 
    lazy: true, 
    computed: true, 
    onTrack, 
    onTrigger, 
    scheduler 
  }); 
  recordInstanceBoundEffect(runner); 
  runner(); 
return () => { 
    stop(runner); 
if (instance) { 
      remove(instance.effects, runner); 
    } 
  }; 
}

可以看到，getter 函数就是对 source 函数的简单封装，它会先判断组件实例是否已经销毁，然后每次执行 source 函数前执行 cleanup 清理函数。

watchEffect 内部创建的 runner 对应的 scheduler 对象就是 scheduler 函数本身，这样它再次执行时，就会执行这个 scheduler 函数，并且传入 runner 函数作为参数，其实就是按照一定的调度方式去执行基于 source 封装的 getter 函数。

创建完 runner 后就立刻执行了 runner，其实就是内部同步执行了基于 source 封装的 getter 函数。

在执行 source 函数的时候，会传入一个 onInvalidate 函数作为参数，接下来我们就来分析它的作用。

注册无效回调函数

有些时候，watchEffect 会注册一个副作用函数，在函数内部可以做一些异步操作，但是当这个 watcher 停止后，如果我们想去对这个异步操作做一些额外事情（比如取消这个异步操作），我们可以通过 onInvalidate 参数注册一个无效函数。

import {ref, watchEffect } from 'vue' 
const id = ref(0) 
watchEffect(onInvalidate => { 
const token = performAsyncOperation(id.value) 
  onInvalidate(() => { 
    token.cancel() 
  }) 
})

我们利用 watchEffect 注册了一个副作用函数，它有一个 onInvalidate 参数。在这个函数内部通过 performAsyncOperation 执行某些异步操作，并且访问了 id 这个响应式对象，然后通过 onInvalidate 注册了一个回调函数。

如果 id 发生变化或者 watcher 停止了，这个回调函数将会执行，然后执行 token.cancel 取消之前的异步操作。

我们来回顾 onInvalidate 在 doWatch 中的实现：

const onInvalidate = (fn) => { 
  cleanup = runner.options.onStop = () => { 
    callWithErrorHandling(fn, instance, 4 ); 
  }; 
};

实际上，当你执行 onInvalidate 的时候，就是注册了一个 cleanup 和 runner 的 onStop 方法，这个方法内部会执行 fn，也就是你注册的无效回调函数。

也就是说当响应式数据发生变化，会执行 cleanup 方法，当 watcher 被停止，会执行 onStop 方法，这两者都会执行注册的无效回调函数 fn。

通过这种方式，Vue.js 就很好地实现了 watcher 注册无效回调函数的需求。

总结

好的，到这里我们这一节的学习也要结束啦，通过这节课的学习，你应该掌握了侦听器内部实现原理，了解侦听器支持的几种配置参数的作用，以及异步任务队列的设计原理。

你也应该掌握侦听器的常见应用场景：如何用 watch API 观测数据的变化去执行一些逻辑，如何利用 watchEffect API 去注册一些副作用函数，如何去注册无效回调函数，以及如何停止一个正在运行的 watcher。

相比于计算属性，侦听器更适合用于在数据变化后执行某段逻辑的场景，而计算属性则用于一个数据依赖另外一些数据计算而来的场景。

最后，给你留一道思考题目，在组件中创建的自定义 watcher，在组件销毁的时候会被销毁吗？是如何做的呢？欢迎你在留言区与我分享。

本节课的相关代码在源代码中的位置如下：
packages/runtime-core/src/apiWatch.ts
packages/runtime-core/src/scheduler.ts