Dive into Vue：调度执行 // Owen

Previously on Dive into Vue: 响应式系统

调度执行机制，指在响应式系统中trigger函数具备的，可以决定副作用执行的时机、次数和方式的能力。在 Vue 的响应式系统中，除了依赖追踪，调度机制也是一个至关重要的组成部分。它决定了副作用在何时、以何种顺序执行，以及如何避免不必要的计算。深入调度执行机制，是理解 Vue 响应式系统不可或缺的步骤。

为什么需要调度机制？

在当前版本的响应式系统视线中，代码不包含调度机制。当触发trigger函数时，会立即执行副作用。考虑下面的例子：

const count = ref(0)

watchEffect(() => {
  console.log(count.value)
})

count.value++
count.value++
count.value++

如果没有调度机制，上述代码会输出：

而在 Vue 中，实际上只会输出：

0
3

即：数据只更新了一次。不难得出结论：在 Vue 中，实际上trigger并不会直接执行副作用，而是将更新任务统一交给了调度机制进行处理。这避免了大量中间无用状态导致的重复更新。

在响应式系统中的代码实现目前还无法做到这一点。在已有逻辑的基础上，可以模仿 Vue 实现简化的调度机制。

支持调度器

当前的任务，是需要trigger可以通过调度机制调用副作用。这就需要修改副作用函数的结构，使其支持传入一个自定义的调度器。

// 副作用函数注册时，允许传入`options`配置项，通过`options.scheduler`指定自定义调度器
type EffectRegisterOptions = {
  scheduler?: (fn: Function) => void
}
// 更新副作用函数的类型，使用`options`存储注册时的配置项
type EffectFn = Function & {
  deps: Deps[]
  options: EffectRegisterOptions
}

function effect(fn: Function, options: EffectRegisterOptions = {}) {
  const effectFn = () => {
    cleanup(effectFn)
    activeEffect = effectFn
    effectStack.push(effectFn)
    fn()
    effectStack.pop()
    activeEffect = effectStack[effectStack.length - 1]
  }

  effectFn.options = options
  effectFn.deps = []
  effectFn()
}

调度器options.scheduler可由用户自定义。那么，如何实现上文中提到的省略中间状态、避免额外更新的调度器呢？可以利用微任务队列在事件循环中「延后执行」的特性。简单的实现代码如下：

// 任务队列
const queue = new Set<Function>()
// 利用 Promise 实现微任务队列
const p = Promise.resolve()
// 是否正在刷新任务队列
let isFlushing = false

function flushJob() {
  if (isFlushing) return
  isFlushing = true
  p.then(() => {
    // 将队列中的任务依次执行
    queue.forEach(job => job())
  }).finally(() => {
    isFlushing = false
  })
}

effect(() => {
  console.log(proxyObj.count)
}, {
  scheduler: (fn) => {
    // 每次调度时，将副作用函数添加到队列中而非立即执行
    queue.add(fn)
    flushJob()
  }
})

这个调度器使用 Set 来实现一个任务队列。当前文中多次调用count.value ++时，实际上只有一次更新操作会被添加到队列中；随后，isFlusing标志保证了flushJob在一个周期内无论被调用多少次，都只会执行一次。当队列执行完毕并最终被清空时，isFlusing标志会被重置为false，下一个周期又会重复以上过程。

该调度器只为演示 Vue 内部「连续多次修改响应式数据但只会触发一次更新」的基本原理，调度器可以由用户自定义来实现复杂的副作用执行策略。Vue 内部实现的调度器思路类似，但更加完善。

接着，在trigger函数中，当需要执行副作用时，如果副作用函数effectFn上指定了调度器，那么调用调度器函数而不是立即执行副作用。

function trigger(target: object, key: PropertyKey) {
  const depsMap = bucket.get(target)
  if (!depsMap) return
  const deps = depsMap.get(key)
  if (!deps) return
  const effectsToRun = new Set<EffectFn>()
  deps.forEach((effectFn) => {
    if (effectFn !== activeEffect) {
      effectsToRun.add(effectFn)
    }
  })
  effectsToRun.forEach((effectFn) => {
    if (effectFn.options.scheduler) {
      effectFn.options.scheduler(effectFn)
    } else {
      effectFn()
    }
  })
}

截至目前，响应式系统已经具备了调度能力，并且还实现了一个行为类似 Vue 的调度器。

侦听器`watch`

上面一节代码实现了计算属性，本节将实现侦听器watch。

回想侦听器的使用方法：

watch(target, (newVal, oldVal) => {
  console.log('count changed from', oldVal, 'to', newVal)
})

可以得出几个特性：

侦听器侦听一个响应式数据或一个 getter；
当侦听的目标发生变化时，执行传入的回调函数；
传入回调时，可接收新值和旧值参数；
可以传入immediate选项，在侦听器创建时立即执行回调；
可以传入flush参数，指定回调函数的执行时机。

其中，第 5 条特性与 Vue 对 DOM 的更新机制相关。前 4 条特性目前可以被实现。

Vue 中，watch的第一个参数可以是响应式数据或一个 getter 函数。首先考虑响应式数据的情况：假设需要侦听的响应式数据是proxyObj，当proxyObj.count被修改为 100 时应当执行回调函数。这天然符合前文中 effect 的调度器的特征（即在响应式数据发生变化时被调用）。因此要实现第 2 条功能，只需要在watch的实现逻辑中，于调度器函数内调用传入的回调函数即可。

function watch(target, cb: () => void) {
  effect(() => target.count, {
    scheduler() {
      cb()
    }
  })
}

watch(proxyObj, () => {
  console.log('count changed')
})

proxyObj.count ++

但是，代码中注册副作用的时候，硬编码了target.count，这使得watch只能侦听count属性。事实上，如果修改了proxyObj的其他属性，我们也期望侦听器的回调可以执行。若要实现这一点，需要对 effect 传入的副作用函数进行封装，递归地访问target中的所有属性：

function watch(target: object, cb: () => void) {
  effect(() => traverse(target), {
    scheduler() {
      cb()
    }
  })
}

function traverse(value: any, seen = new Set<any>) {
  // 基本数据类型或已经被访问过，则不递归
  if (typeof value !== 'object' || value === null || seen.has(value)) return
  seen.add(value)
  // 暂不考虑数组等其他数据结构
  // value 作为普通对象，递归访问其所有属性
  for (const k in value) {
    traverse(value[k], seen)
  }
  return value
}

接下来考虑传入的target是一个 getter 函数的情况。如果用户直接传入一个 getter，那么就不需要在effect中调用traverse函数，因为 getter 函数内部已经包含了对响应式数据的访问逻辑。此时需要增加一层判断：

function watch(target: object | Function, cb: () => void) {
  let getter: Function;
  if (typeof target === 'function') {
    getter = target
  } else {
    getter = () => traverse(target)
  }
  effect(() => getter(), {
    scheduler() {
      cb()
    }
  })
}

支持回调参数

在第 3 条功能特性中，侦听器的回调函数支持包含两个参数，分别是侦听对象发生变化后的新值和变化前的旧值，这需要能够在执行 getter 的时候保存其返回值。在实现计算属性时，已经实现了类似的缓存功能，通过启用lazy来实现手动调用获取返回值。在此处，可以使用类似的思路：

function watch(target: object | Function, cb: (newVal?: any, oldVal?: any) => void) {
  let getter: Function;
  if (typeof target === 'function') {
    getter = target
  } else {
    getter = () => traverse(target)
  }
  let newVal: any, oldVal: any
  const effectFn = effect(() => getter(), {
    lazy: true,
    scheduler() {
      // 副作用执行时，产生新值
      newVal = effectFn()
      // 调用回调函数
      cb(newVal, oldVal)
      // 更新旧值
      oldVal = newVal
    }
  })

  // 手动调用获取旧值
  oldVal = effectFn()
}

watch(() => proxyObj.count, (newVal, oldVal) => {
  console.log('count changed from', oldVal, 'to', newVal)
})

在上面的代码中，已知调度器会在侦听对象发生变化时被调用，因此可以在调度器中执行effectFn获取响应式数据变化后的新值。在调用watch时手动执行的effectFn发生于完成依赖收集后，此时响应式数据没有发生变化，获取到的就是初始值。此后，每次触发cb后，都更新旧值，避免在下次触发cb时得到到错误的旧值。

立即执行的`watch`

在 Vue 中，立即执行的watch指在创建侦听器时立即执行一次回调函数。这个功能相对比较简单，只需要在watch的实现逻辑中，增加一个判断：如果immediate选项为true，则在创建侦听器时手动调用一次回调函数即可。

type WatchOptions = {
  immediate?: boolean
}

function watch(
  target: object | Function,
  cb: (newVal?: any, oldVal?: any) => void,
  options: WatchOptions = {}
) {
  let getter: Function;
  if (typeof target === 'function') {
    getter = target
  } else {
    getter = () => traverse(target)
  }
  let newVal: any, oldVal: any

  // 将调度器函数封装为 job
  const job = () => {
    newVal = effectFn()
    cb(newVal, oldVal)
    oldVal = newVal
  }

  const effectFn = effect(() => getter(), {
    lazy: true,
    scheduler: job
  })

  // 如果传入了 immediate 选项，立即调用 job 函数
  if (options.immediate) {
    job()
  } else {
    oldVal = effectFn()
  }
}

值得注意的一点是，设置了立即执行并调用job()时，旧值应当是undefined。

目前为止的完整代码

响应式系统核心：

// 全局存储桶，保存每个对象的依赖关系
type Bucket = WeakMap<object, DepsMap>
// 响应式对象中，每个属性和其对应的副作用函数集合的依赖关系表
type DepsMap = Map<PropertyKey, Deps>
// 每个属性的副作用函数集合
type Deps = Set<EffectFn>
// 副作用函数注册时，允许传入`options`配置项，通过`options.scheduler`指定自定义调度器
type EffectRegisterOptions = {
  scheduler?: (fn: Function) => void
  lazy?: boolean
}
// 更新副作用函数的类型，使用`options`存储注册时的配置项
type EffectFn = Function & {
  deps: Deps[]
  options: EffectRegisterOptions
}

const obj: Record<PropertyKey, any> = {
  firstName: 'John',
  lastName: 'Doe'
}

const bucket: Bucket = new WeakMap()
let activeEffect: EffectFn
const effectStack: EffectFn[] = []

function cleanup(effectFn: EffectFn) {
  for (let i = 0; i < effectFn.deps.length; i++) {
    const deps = effectFn.deps[i]
    deps.delete(effectFn)
  }
  effectFn.deps.length = 0
}

function effect(fn: Function, options: EffectRegisterOptions = {}) {
  const effectFn: EffectFn = () => {
    cleanup(effectFn)
    activeEffect = effectFn
    effectStack.push(effectFn)
    const res = fn()
    effectStack.pop()
    activeEffect = effectStack[effectStack.length - 1]
    return res
  }
  effectFn.deps = []
  effectFn.options = options
  if (!options.lazy) {
    effectFn()
  }
  return effectFn
}

function track(target: object, key: PropertyKey) {
  if (!activeEffect) return
  let depsMap = bucket.get(target)
  if (!depsMap) {
    bucket.set(target, (depsMap = new Map()))
  }
  let deps = depsMap.get(key)
  if (!deps) {
    depsMap.set(key, (deps = new Set()))
  }
  deps.add(activeEffect)
  console.log(activeEffect)
  // 错误：activeEffect.deps 是 undefined，activeEffect 是一个函数，并非预期的`EffectFn`类型
  activeEffect.deps.push(deps)
}

function trigger(target: object, key: PropertyKey): void {
  const depsMap = bucket.get(target)
  if (!depsMap) return
  const deps = depsMap.get(key)
  if (!deps) return
  const effectsToRun = new Set<EffectFn>()
  deps.forEach((effectFn) => {
    if (effectFn !== activeEffect) {
      effectsToRun.add(effectFn)
    }
  })
    effectsToRun.forEach((effectFn) => {
    if (effectFn.options.scheduler) {
      effectFn.options.scheduler(effectFn)
    } else {
      effectFn()
    }
  })
}

const proxyObj = new Proxy(obj, {
  get: (target, key: PropertyKey) => {
    track(target, key)
    return target[key]
  },
  set: (target, key: PropertyKey, value) => {
    target[key] = value
    trigger(target, key)
    return true
  }
})

计算属性：

function computed(getter: Function) {
  let dirty = true
  let cache: any = undefined

  const effectFn = effect(() => {
    console.log('effectFn invoked')
    return getter()
  }, {
    lazy: true,
    scheduler() {
      dirty = true
      trigger(obj, 'value')
    }
  })
  
  const obj = {
    get value() {
      if (dirty) {
        cache = effectFn()
        dirty = false
      }
      track(obj, 'value')
      return cache
    }
  }

  return obj
}

侦听器：

type WatchOptions = {
  immediate?: boolean
}

function watch(
  target: object | Function,
  cb: (newVal?: any, oldVal?: any) => void,
  options: WatchOptions = {}
) {
  let getter: Function;
  if (typeof target === 'function') {
    getter = target
  } else {
    getter = () => traverse(target)
  }
  let newVal: any, oldVal: any

  const job = () => {
    newVal = effectFn()
    cb(newVal, oldVal)
    oldVal = newVal
  }

  const effectFn = effect(() => getter(), {
    lazy: true,
    scheduler: job
  })

  if (options.immediate) {
    job()
  } else {
    oldVal = effectFn()
  }
}

function traverse(value: any, seen = new Set<any>) {
  if (typeof value !== 'object' || value === null || seen.has(value)) return
  seen.add(value)
  for (const k in value) {
    traverse(value[k], seen)
  }

  return value
}

Fin.

cd ../

Dive into Vue：调度执行

为什么需要调度机制？

支持调度器

计算属性`computed`

Lazy Effect

缓存计算结果

嵌套的副作用

侦听器`watch`

侦听对象，执行回调

支持回调参数

立即执行的`watch`

目前为止的完整代码