Vue3源码分析调度器与watch用法原理

本文主要内容
调度器
1.添加任务(queueJobs)
2.二分法找到插入位置(findInsertionIndex)
3.将执行任务的函数推入微任务队列(queueFlush)
4.执行普通任务(flushJobs)
5.添加后置任务(queuePostFlushCb)
6.queuePostRenderEffect
7.执行后置队列任务(flushPostFlushJobs)
8.执行前置任务队列(flushPreFlushCbs)
9.nextTick
调度器总结
watch用法
选项式watch Api的实现
创建watch对象(createWatchr)
选项式watch Api总结
函数式watch的实现(下面统称watch)
1.watch
2.doWatch
watch总结

本文主要内容

学习Vue3的调度器原理。
了解nextTick的实现、为何在nextTick中可以获取到修改后的DOM属性。
pre、post、和普通任务的执行过程。
watch的实现原理。

调度器

1.添加任务(queueJobs)

调度器想要运转需要添加任务到调度器队列当中,我们需要知道Vue调度器队列一共有两种,分别为queue、pendingPostFlushCbs。
queue:装载前置任务和普通任务的队列。
pendingPostFlushCbs:装载后置任务的队列。

下面我们来看看对于前置任务和普通任务添加到queue中的函数queueJobs。

 
//递归:当前父亲正在执行一个任务,在执行任务
//期间又添加了一个新的任务,这个新的任务与当前
//执行的任务是同一个任务,跳过去重的检验
//如果不允许递归,那么任务不会被添加到队列中
function queueJob(job) {
  //job自身允许递归,那么跳过去重检查(只跳过当前执行任务的去重检查)
  if (
    !queue.length ||
    !queue.includes(
      job,
      isFlushing && job.allowRecurse ? flushIndex + : flushIndex
    )
  ) {
    //如果任务没有id代表没有优先级
    //放到任务队列的最后面
    if (job.id == null) {
      queue.push(job);
    }
    //利用二分法找到任务优先级需要插入的位置
    else {
      queue.splice(findInsertionIndex(job.id),, job);
    }
    //执行任务
    queueFlush();
  }
}

这里我们需要知道一个概念-->递归,这里的递归是指:当前正在执行的任务和需要添加的任务是同一个任务,如果设置了需要递归(job.allowRecurse=true)那么就允许这个任务进入queue队列中,否则不允许进入。
job:我们还需要知道一个任务的格式。首先job必须是一个函数,他还可以具有以下属性。

 
const job = function(){}
job.id:Number //用于设置当前任务的优先级越小的值优先级越高。
job.allowRecurse:Boolean //是否允许递归。
job.pre:Boolean //用于判断是否是前置任务。
job.active:Boolean //当前任务是否可以执行。为false在执行阶段跳过执行。

queueJobs执行流程:根据任务的id(优先级)利用二分法找到需要插入的位置,插入到queue队列当中,调用queueFlush推入执行任务的函数到微任务队列。

2.二分法找到插入位置(findInsertionIndex)

这个函数比较简单,大家看看代码就可以啦！

 
//找到插入的位置
//例如[,2,3,8,9,10,100]
//当前插入的id为
//插入后应该为[,2,3,8,9,10,20,100]
//也就是说最终返回的start=
//插入流程解析:
//.假设当前执行到第二个任务即flushIndex=2
//那么start =;end = 7;middle=4;
//    middleJobId=;9<20 start=5;
//继续循环:middle=;middleJobId=100;end=6
//结束循环start =;这就是需要插入的位置
function findInsertionIndex(id) {
  let start = flushIndex +;
  let end = queue.length;
  while (start < end) {
    //>>>1=>100 8=>4
    //>>>1=>110 12=>6
    //>>>1=>101 10=>5
    //>>>1=>100 9=>4
    //计算出中间值,向下取整
    const middle = (start + end) >>>;
    //获取job的id
    const middleJobId = getId(queue[middle]);
    middleJobId < id ? (start = middle +) : (end = middle);
  }
  return start;
}
//获取当前任务的id
const getId = (job) => (job.id == null ? Infinity : job.id);

3.将执行任务的函数推入微任务队列(queueFlush)

 
function queueFlush() {
  //当前没有执行任务且没有任务可执行
  if (!isFlushing &amp;&amp; !isFlushPending) {
    //等待任务执行
    isFlushPending = true;
    //将flushJobs放入微任务队列
    currentFlushPromise = resolvedPromise.then(flushJobs);
  }
}

isFlushing:判断当前是否正在执行任务。
isFlushPending:判断当前是否有等待任务,任务的执行是一个微任务,它将会被放到微任务队列,那么对于渲染主线程来说,当前还没有执行这个微任务,在执行这个微任务之前都属于等待阶段。
queueFlush执行流程:判断当前是否没有执行任务、且任务队列当中没有任务,如果是那么设置当前为等待阶段。最后将flushJobs(执行任务的函数)推入微任务队列。

4.执行普通任务(flushJobs)

 
function flushJobs(seen) {
  isFlushPending = false; //当前不是等待状态
  isFlushing = true; //当前正在执行任务
  seen = seen || new Map();
  //原文译文:
  //在flush之前对queue排序这样做是为了:
  //.组件更新是重父组件到子组件(因为父组件总是在子组件之前创建
  //所以父组件的render副作用将会有更低的优先级
  //.如果子组件在父组件更新期间并未挂载,那么可以跳过
  queue.sort(comparator);
  //监测当前任务是否已经超过了最大递归层数
  const check = (job) => checkRecursiveUpdates(seen, job);
  try {
    for (flushIndex =; flushIndex < queue.length; flushIndex++) {
      const job = queue[flushIndex];
      if (job && job.active !== false) {
        if (check(job)) {
          continue;
        }
        callWithErrorHandling(job, null,);
      }
    }
  } finally {
    //执行完所有的任务之后,初始化queue
    //调用post任务,这些任务调用完折后
    //可能在执行这些任务的途中还有新的
    //任务加入所以需要继续执行flushJobs
    flushIndex =;
    queue.length =;
    flushPostFlushCbs(seen);
    isFlushing = false;
    currentFlushPromise = null;
    if (queue.length || pendingPostFlushCbs.length) {
      flushJobs(seen);
    }
  }
}

seen:这是一个Map,用于缓存job的执行次数,如果超过了RECURSION_LIMIT的执行次数,将会警用户。
RECURSION_LIMIT:Vue默认值为100。这个值不可以让用户修改(常量值)。
flushJobs执行流程:获取queue队列中的每一个任务,检测这个任务是否嵌套执行了100次以上,超过了则警告用户。然后执行当前任务直到flushIndex === queue.length。(queue的长度可能会持续增加)。调用flushPostFlushCbs执行后置队列的任务。
由于在执行后置队列任务的时候可能又向queue中添加了新的任务,那么就需要执行完后置队列后再调用flushJobs。

5.添加后置任务(queuePostFlushCb)

 
function queuePostFlushCb(cb) {
  if (!shared.isArray(cb)) {
    if (
      !activePostFlushCbs ||
      !activePostFlushCbs.includes(
        cb,
        cb.allowRecurse ? postFlushIndex + : postFlushIndex
      )
    ) {
      pendingPostFlushCbs.push(cb);
    }
  } else {
    pendingPostFlushCbs.push(...cb);
  }
  queueFlush();
}

与添加普通任务到队列中一样,添加完成后调用queueFlush开启调度。

6.queuePostRenderEffect

 
function queueEffectWithSuspense(fn, suspense) {
  //对suspense的处理,暂时不详细解释
  if (suspense && suspense.pendingBranch) {
    if (shared.isArray(fn)) {
      suspense.effects.push(...fn);
    } else {
      suspense.effects.push(fn);
    }
  } else {
    //如果是普通的任务则放入后置队列
    queuePostFlushCb(fn);
  }
}

如果传递了suspense那么调用suspense的api。
没有传递suspense当作一般的后置任务即可。

7.执行后置队列任务(flushPostFlushJobs)

 
function flushPostFlushCbs(seen) {
  if (pendingPostFlushCbs.length) {
    //克隆等待执行的pendingPost
    const deduped = [...new Set(pendingPostFlushCbs)];
    pendingPostFlushCbs.length =; //设置为0
    //当前函数是后置队列的任务发起的,那么不能
    //直接运行任务,而是将任务放到avtivePostFlushCbs任务之后
    if (activePostFlushCbs) {
      activePostFlushCbs.push(...deduped);
      return;
    }
    activePostFlushCbs = deduped;
    seen = seen || new Map();
    //排序(post依然有优先级)
    activePostFlushCbs.sort((a, b) => getId(a) - getId(b));
    for (
      postFlushIndex =;
      postFlushIndex < activePostFlushCbs.length;
      postFlushIndex++
    ) {
      //检测执行深度
      if (checkRecursiveUpdates(seen, activePostFlushCbs[postFlushIndex])) {
        continue;
      }
      //调用这个postJob
      activePostFlushCbs[postFlushIndex]();
    }
    //初始化
    activePostFlushCbs = null;
    postFlushIndex =;
  }
}

flushPostFlushCbs执行流程:和flushJobs差不多,拿到pendingPostFlushCbs队列中的任务并执行他们,在执行完成后初始化postFulshIndex指针。
之所以后置队列一定会在完成普通任务和前置任务后执行,是因为无论你是通过queueJobs添加任务发起调度还是通过queuePostFlushCb添加任务发起调度,都总是调用flushJobs,而在flushJobs的实现中,总是先清空queue队列在执行pendingPostFlushCbs。
activePostFlushCbs作用:想象一个场景,如果我直接通过调用flushPostFlushJobs发起调度那么任务将不会是异步的,并且会打乱调度器的执行顺序,所以有了这个属性。若当前已经存在了activePostFlushCbs表示正在执行后置队列的任务,在任务中调用flushPostFlushJobs并不会直接执行,而是会把pendingPostFlushcbs中的任务放到avtivePostFlushCbs任务的后面。这样就保证了调度器的顺序执行。

8.执行前置任务队列(flushPreFlushCbs)

 
function flushPreFlushCbs(seen, i = isFlushing ? flushIndex + : 0) {
  seen = seen || new Map();
  for (; i < queue.length; i++) {
    const cb = queue[i];
    if (cb && cb.pre) {
      if (checkRecursiveUpdates(seen, cb)) {
        continue;
      }
      queue.splice(i,);
      i--;
      cb();
    }
  }
}

添加前置任务的方法:对添加的任务函数Job添加pre属性。


job.pre = true

这里需要注意,对于前置任务和普通任务都会被添加到queue当中,如果调用的flushJobs触发任务执行,那么前置任务和普通任务都会被执行。他们的执行顺序为高优先级的先执行(id小的先执行)。相同优先级的前置任务先执行。
flushPreFlushCbs执行流程:在queue中找到带有pre属性的任务,执行并在queue中删除这个任务。
对于处于执行后置任务的状态,同时调用了flushPostFlushCbs发起后置任务的调度,那么会将新增的任务加到activePostFlushCbs中。但是对于前置任务是不需要这么做的,如果通过调用flushPreFlushCbs发起调度那么前置任务将会是同步执行。我们来看这样一个例子。

 
function a(){
  console.log()
}
function b(){
  console.log()
}
a.pre = true
queueJobs(a)
queueJobs(b)
flushPreFlushCbs()
//打印: 111

如何理解呢？首先a任务是前置任务,a、b任务都被添加到了queue队列中,同时发起了调度,但是这是一个微任务,而当前执行的任务还未执行完成,所以会先调用flushPreFlushCbs。那么就会调用前置任务也就是a任务。调用完成后删除queue队列中的a任务,此时queue队列中只有b任务了。然后执行微任务,进一步调用b任务。

9.nextTick

场景:在修改了响应式数据后,想要获取到最新DOM上的数据,因为只修改了相应式数据,目前DOM还未发生改变所以获取不到改变后的DOM属性。

 
<script> 
import { nextTick } from 'vue' 
export default {  
  data() {   
    return { count: } }, 
    methods: { 
      async increment() { 
        this.count++ // DOM 还未更新
        //
        console.log(document.getElementById('counter').textContent)
        await nextTick() // DOM 此时已经更新 
        console.log(document.getElementById('counter').textContent) // 
        } 
     } 
 } 
 </script>
 <template> 
   <button id="counter" @click="increment">{{ count }}</button> 
 </template>

nextTick实现:

 
function nextTick(fn) {
  const p = currentFlushPromise || resolvedPromise;
  return fn ? p.then(this ? fn.bind(this) : fn) : p;
}

currentFlushPromise:在调用queueFlush时会创建一个微任务,将flushJobs推入微任务队列。

 
function queueFlush() {
  if (!isFlushing && !isFlushPending) {
    isFlushPending = true;
    currentFlushPromise = resolvedPromise.then(flushJobs);
  }
}

resolvedPromise:状态为fulfilled的Promise。
如果当前队列中没有任务则p=resolvedPromise,直接将fn推入微任务队列。因为调度器队列中无任务所以不存在DOM的更新。
如果当前队列中有任务则p=currentFlushPromise,若当前正在执行flushJobs那么currentFlushPromise的状态为fulfilled则会将fn推入微任务队列,当然前提是flushJobs已经执行完才有可能执行fn,而只要flushJobs执行完毕DOM也已经完成了更新。若当前没有执行flushJobs,那么currentFlushPromise的状态为pending,就不可能将fn推入微任务队列。综上就保证了fn一定在DOM更新后触发。

调度器总结

调度器的调度队列分为后置队列和普通队列。
普通队列中包含了前置任务和普通任务。如果通过flushPreFlushCbs调用那么前置任务为同步任务。执行完成后删除普通队列中相对应的任务。如果通过flushJobs调用,那么调用顺序按照优先级高低排列,相同优先级的前置任务先调用。
后置队列任务一定在普通队列清空后执行。
普通任务和后置任务为异步,前置任务可能为同步可能为异步。
在将任务放入队列当中时就已经自动发起了调度,用户可以不通过手动调用。如果手动调用flushPostFlushCbs实际上是将任务放到队列中,而不是重新开启调度。

watch用法

选项式

 
<script>
export default {
  watch{
    a(){},
    b:"meth"//在methods中声明的方法
    c:{
      handler(val,oldVal){},
      deep:true,//开启深度监视
      immediate:true//立即调用handler
    },
    "d.a":function(){}
  }
}
</script>

函数式

 
const callback = ([aOldVal,aVal],[bOldVal,bVal])=>{}
//监听源 监听源发生改变的回调函数 选项
watch(["a","b"], callback, {
  flush: 'post',
  onTrack(e) {
    debugger
  },
  deep:true,
  immediate:true,
})

选项式watch Api的实现

 
//这一段代码在Vue源码分析(7)中出现过
//不了解的可以看看上一篇文章
//对每一个watch选项添加watcher
if (watchOptions) {
  for (const key in watchOptions) {
    createWatcher(watchOptions[key], ctx, publicThis, key);
  }
}

这里的watchOptions就是用户写的选项式api的watch对象。

创建watch对象(createWatchr)

 
function createWatcher(raw, ctx, publicThis, key) {
  //可以监听深度数据例如a.b.c
  const getter = key.includes(".")
    ? createPathGetter(publicThis, key)
    : () => publicThis[key];
  //raw可以是字符串,会读取methods中的方法
  if (shared.isString(raw)) {
    const handler = ctx[raw];
    if (shared.isFunction(handler)) {
      //进行监听
      watch(getter, handler);
    } else {
      warn(`Invalid watch handler specified by key "${raw}"`, handler);
    }
  }
  //如果是函数 监听
  else if (shared.isFunction(raw)) {
    watch(getter, raw.bind(publicThis));
  }
  //如果是对象
  else if (shared.isObject(raw)) {
    //数组遍历,获取每一个监听器在执行createWatcher
    if (shared.isArray(raw)) {
      raw.forEach((r) => createWatcher(r, ctx, publicThis, key));
    }
    //对象
    else {
      //handler可能是字符串重ctx上获取
      //也可能是函数
      //获取到handler后调用watch
      const handler = shared.isFunction(raw.handler)
        ? raw.handler.bind(publicThis)
        : ctx[raw.handler];
      if (shared.isFunction(handler)) {
        watch(getter, handler, raw);
      } else {
        warn(
          `Invalid watch handler specified by key "${raw.handler}"`,
          handler
        );
      }
    }
  } else {
    warn(`Invalid watch option: "${key}"`, raw);
  }
}

选项式watch的键可以是"a.b.c"这样的形式也可以是普通的"a"形式,它的值可以是字符串,函数,对象,数组。此函数主要对不同形式的参数做重载。最终都是调用watch函数。
对于键为"a.b"形式的需要调用createPathGetter创建一个getter函数,getter函数返回"a.b"的值。
对于值为字符串的我们需要从methods中获取对应的方法。因为之前许多重要属性都代理到ctx上了所以只需要访问ctx即可。
对于值为函数的我们只需要将key作为watch的第一个参数,值作为watch的第二个参数即可。
对于值为对象的获取handler作为watch第二个参数,将raw作为第三个参数(选项)传入watch即可。
对于值为数组的,表示需要开启多个监听,遍历数组递归调用createWatcher即可。

选项式watch Api总结

对于选项式watch Api本质上还是调用的函数式watch Api进行实现的。这里只是做了重载,对于不同的配置传递不同的参数给watch。所以接下来我们重点分析函数式watch Api的实现。

函数式watch的实现(下面统称watch)

1.watch

 
function watch(source, cb, options) {
  //cb必须是函数
  if (!shared.isFunction(cb)) {
    console.warn();
  }
  return doWatch(source, cb, options);
}

source:监听源,可以是数组(代表监听多个变量)。
cb:监听源发生改变时,调用的回调函数。
options:watch函数的可选项。
如果传递的cb不是函数需要警告用户,这可能导致错误。

2.doWatch

这个函数非常长,也是watch的实现核心,我们分多个部分讲解。
大致原理:收集source中响应式元素包装成getter,在new ReactiveEffect中传递调用run方法执行getter就会收集到依赖,然后当触发依赖更新的时候就会调用scheduler,在根据flush参数,选择同步执行scheduler还是加入调度器。

 
function doWatch(
  source, //getter ()=>[监听的数据]
  cb, //回调函数
  //获取当前watch的选项
  { immediate, deep, flush, onTrack, onTrigger } = shared.EMPTY_OBJ
) {
  //immediate和deep属性必须有cb
  if (!cb) {
    if (immediate !== undefined) {
      warn(
        `watch() "immediate" option is only respected when using the ` +
          `watch(source, callback, options?) signature.`
      );
    }
    if (deep !== undefined) {
      warn(
        `watch() "deep" option is only respected when using the ` +
          `watch(source, callback, options?) signature.`
      );
    }
  }
  //省略第二部分代码
 }

第一部分的代码主要是检测参数。对于没有cb参数但是又有immediate和deep选项的需要警告用户。

 
//获取当前实例
const instance = getCurrentInstance();
let getter;
let forceTrigger = false; //强制触发
let isMultiSource = false; //是否多个数据
//判断监听的数据是否是ref
if (reactivity.isRef(source)) {
  getter = () => source.value;
  forceTrigger = reactivity.isShallow(source);
}
//判断数据是否是响应式
else if (reactivity.isReactive(source)) {
  getter = () => source;
  deep = true;
}
//判断数据是否是数组
else if (shared.isArray(source)) {
  isMultiSource = true;
  //source中有一个是响应式的
  //就需要触发
  forceTrigger = source.some(
    (s) => reactivity.isReactive(s) || reactivity.isShallow(s)
  );
  //()=>[proxy,()=>proxy,ref]
  getter = () =>
    source.map((s) => {
      if (reactivity.isRef(s)) {
        return s.value;
      } else if (reactivity.isReactive(s)) {
        //遍历响应式对象s 这个getter会作为ReactiveEffect的
        //第一个参数,在调用run的时候遍历所有的值
        //确保能让每一个变量都能收集到effect
        return traverse(s);
      }
      //调用监听的函数
      else if (shared.isFunction(s)) {
        return callWithErrorHandling(s, instance,);
      } else {
        //提示非法source信息
        warnInvalidSource(s);
      }
    });
}
//省略第三部分代码

如果监听的数据是ref类型,包装成getter形式。
如果监听的数据是reactive类型,需要设置为深度监听。
如果监听的数据是数组,设置变量isMultiSource=true表示当前监听了多个变量,同时判断监听的所有数据中是否有相应式对象,如果有就必须强制触发。设置getter。
我们可以发现所有的监听数据源都会被包装成getter,这是因为底层都是调用reactivity库的watchEffect,而第一个参数必须是函数,当调用这个函数访问到的变量都会收集依赖。所以如果当前元素为reactive元素的时候需要遍历这个元素的所有值以便所有的变量都能收集到对应的依赖。

 
//()=>[proxy]传入的是一个函数
else if (shared.isFunction(source)) {
  if (cb) {
    //让getter为这个函数
    getter = () => callWithErrorHandling(source, instance,);
  } else {
    //如果没有回调函数
    getter = () => {
      if (instance && instance.isUnmounted) {
        return;
      }
      if (cleanup) {
        cleanup();
      }
      return callWithAsyncErrorHandling(source, instance,, [onCleanup]);
    };
  }
}
//省略第四部分代码

如果监听的数据是函数,先判断是否有cb,如果有cb则将监听源函数作为getter。
如果没有传递cb,那么这个函数将会作为getter和回调函数cb。
我们来详细说说cleanup的作用。先来看看官方的测试用例:

 
watch(async (onCleanup) => {
  const { response, cancel } = doAsyncWork(id.value)
  // `cancel` 会在 `id` 更改时调用
  // 以便取消之前
  // 未完成的请求
  onCleanup(cancel)
  data.value = await response
})

它被用来做副作用清除。第一次调用getter的时候是作为收集依赖,所以cleanup为空不执行,然后调用source函数,在这个函数中会收到onCleanup的参数,如果你在source函数中调用了onCleanup函数那么cleanup将会被赋值。当id发生改变之后再次调用getter函数(此时作为cb),这时候cleanup就会被调用,也就是官方说的cancle函数会在id更改时调用。
我们继续第四部分代码的分析:

 
//不是以上情况,让getter为空函数
else {
  getter = shared.NOOP;
  //警告
  warnInvalidSource(source);
}
//省略第五部分代码

这表示没有需要监听的数据源,将getter设置为空函数,同时警告用户。

 
const INITIAL_WATCHER_VALUE = {}
//getter作为参数传入ReactiveEffect
//调用run的时候会调用getter,确保
//所有的属性都能够收集到依赖
if (cb && deep) {
  const baseGetter = getter;
  getter = () => traverse(baseGetter());
}
let cleanup;
//调用effect.stop的时候触发这个函数
let onCleanup = (fn) => {
  cleanup = effect.onStop = () => {
    callWithErrorHandling(fn, instance,);
  };
};
let oldValue = isMultiSource ? [] : INITIAL_WATCHER_VALUE;
//省略第六部分代码

对于含有deep属性的需要深度遍历,只要在getter中访问了所有变量的值那么这些值都会收集到依赖。
接下来便是onCleanup的实现,大家可以按照上面我说的进行理解。
我们知道在watch可以监听多个数据,那么对应的cb回调函数的参数要收集到这些改变的值。所以如果监听了多个数据源那么oldValue会被设置为数组否则为对象。

 
//回调函数
 const job = () => {
  if (!effect.active) {
    return;
  }
  //传递了cb函数
  if (cb) {
    //watch([a,b],()=>{})
    //newValue=[a,b]
    const newValue = effect.run();
    //未设置deep属性的
    //旧值和新值要发生改变才会调用cb回调函数
    if (
      deep ||
      forceTrigger ||
      (isMultiSource
        ? newValue.some((v, i) => shared.hasChanged(v, oldValue[i]))
        : shared.hasChanged(newValue, oldValue))
    ) {
      //这里的作用上面我们已经讲过了,不在赘述。
      if (cleanup) {
        cleanup();
      }
      callWithAsyncErrorHandling(cb, instance,, [
        newValue,
        oldValue === INITIAL_WATCHER_VALUE ? undefined : oldValue,
        onCleanup,
      ]);
      oldValue = newValue;
    }
  } else {
    //没有cb就只调用getter函数(watchEffect)
    effect.run();
  }
};
//省略第七部分代码

这个job代表的是要传递给Vue调度器的任务,所以这是在创建一个调度器任务。
同时还需要注意这个job是监听的变量发生了改变后才会调用。
这里的effect代表的是ReactiveEffect类的实例,如果还不了解这个类的请阅读Vue3源码分析(2)。
如果没有传递cb那么会调用effect.run()这个函数会去执行getter函数。因为没有传递cb所以回调函数就是getter函数。
如果存在cb,那么会先调用getter函数获取最新的value,然后再调用cb,所以不太建议自己将第一个参数写成函数,这样改变值的时候会调用getter和cb两个函数,如果你在getter中写了副作用那么就会多次调用。
同样cleanup用于清除副作用这里就不再赘述了。

 
//只要有cb则允许递归
job.allowRecurse = !!cb;
let scheduler;
//设置了sync则同步调度,不放入queue进行异步调度(同步)
if (flush === "sync") {
  scheduler = job;
}
//设置了post放到DOM渲染之后执行(异步)
else if (flush === "post") {
  scheduler = () => queuePostRenderEffect(job, instance && instance.suspense);
}
//默认值为pre,放入到queue中执行(异步)
//带有pre的会在DOM渲染前执行
else {
  job.pre = true;
  //给当前的job设置优先级
  if (instance) job.id = instance.uid;
    scheduler = () => queueJob(job);
}
//省略第八部分代码

当监视的数据发生改变的时候会调用job任务,但是job任务是异步调用还是同步调用是可以通过flush参数改变的。
当flush为sync的时候:会同步的执行job任务。
当flush为post的时候:会将job任务推入后置任务队列,也就是会等queue队列任务执行完成之后执行。
当flush为pre的时候:会将job任务设置为前置任务,在调用flushPreFlushCbs的时候执行。执行完成后删除这个任务。当然如果一直不调用flushPreFlushCbs,将会作为普通任务执行,这时候就是异步的了。
最终getter和scheduler都得到了。他们会作为reactiveEffect类的两个参数。第一个为监听的getter函数,在这里面访问的值都会收集到依赖,当这些监听的值发生改变的时候就会调用schgeduler。

 
const effect = new reactivity.ReactiveEffect(getter, scheduler);
//将用户传递的onTrack和onTrigger赋值到effect上
//便于在track和trigger的时候调用
effect.onTrack = onTrack;
effect.onTrigger = onTrigger;
//省略第九部分代码

onTrack:是reactivity库实现的api。当被追踪的时候调用这个函数。
onTrigger:当监视的变量改变的时候触发的函数。
创建ReactiveEffect实例对象,对变量进行监视。

 
//调用了watch之后
//需要立刻执行getter,处理不同的flush参数
if (cb) {
  if (immediate) {
    //有immediate参数立即执行job
    job();
  }
  //否则就只收集依赖调用getter函数
  //并且获取监听的变量
  else {
    oldValue = effect.run();
  }
}
//flush为post需要将收集依赖函数getter
//放到postQueue中
else if (flush === "post") {
  queuePostRenderEffect(
    effect.run.bind(effect),
    instance && instance.suspense
  );
}
//没有设置则收集依赖
else {
  effect.run();
}
//省略第十部分代码

如果含有immediate参数则需要立刻执行job任务,否则调用effect.run()方法(调用getter)收集依赖。
如果flush设置为post那么收集依赖的操作也需要移动到后置队列当中。

 
//watch的停止函数,调用后不再依赖更新
return () => {
  effect.stop();
};

watch会返回一个方法用于取消监听。

watch总结

为了兼容选项式watch处理了不同的配置选项最终调用函数式的watch来实现的监视效果。
watch拥有三个参数:source、cb、options。
source是监听源,可以传递函数,值,数组。但是最后都是包装成getter函数。实现的理念就是通过调用getter函数,访问响应式变量收集依赖,当响应式数据发生改变的时候调用cb。
options中比较重要的配置是flush,他决定了何时收集依赖和触发依赖。当flush为post的时候需要知道收集依赖和触发依赖都将会推入到后置队列当中(DOM更新后触发)。

	//递归:当前父亲正在执行一个任务,在执行任务
	//期间又添加了一个新的任务,这个新的任务与当前
	//执行的任务是同一个任务,跳过去重的检验
	//如果不允许递归,那么任务不会被添加到队列中
	function queueJob(job) {
	//job自身允许递归,那么跳过去重检查(只跳过当前执行任务的去重检查)
	if (
	!queue.length \|\|
	!queue.includes(
	job,
	isFlushing && job.allowRecurse ? flushIndex + : flushIndex
	)
	) {
	//如果任务没有id代表没有优先级
	//放到任务队列的最后面
	if (job.id == null) {
	queue.push(job);
	}
	//利用二分法找到任务优先级需要插入的位置
	else {
	queue.splice(findInsertionIndex(job.id),, job);
	}
	//执行任务
	queueFlush();
	}
	}

	const job = function(){}
	job.id:Number //用于设置当前任务的优先级越小的值优先级越高。
	job.allowRecurse:Boolean //是否允许递归。
	job.pre:Boolean //用于判断是否是前置任务。
	job.active:Boolean //当前任务是否可以执行。为false在执行阶段跳过执行。

	//找到插入的位置
	//例如[,2,3,8,9,10,100]
	//当前插入的id为
	//插入后应该为[,2,3,8,9,10,20,100]
	//也就是说最终返回的start=
	//插入流程解析:
	//.假设当前执行到第二个任务即flushIndex=2
	//那么start =;end = 7;middle=4;
	// middleJobId=;9<20 start=5;
	//继续循环:middle=;middleJobId=100;end=6
	//结束循环start =;这就是需要插入的位置
	function findInsertionIndex(id) {
	let start = flushIndex +;
	let end = queue.length;
	while (start < end) {
	//>>>1=>100 8=>4
	//>>>1=>110 12=>6
	//>>>1=>101 10=>5
	//>>>1=>100 9=>4
	//计算出中间值,向下取整
	const middle = (start + end) >>>;
	//获取job的id
	const middleJobId = getId(queue[middle]);
	middleJobId < id ? (start = middle +) : (end = middle);
	}
	return start;
	}
	//获取当前任务的id
	const getId = (job) => (job.id == null ? Infinity : job.id);

	function queueFlush() {
	//当前没有执行任务且没有任务可执行
	if (!isFlushing && !isFlushPending) {
	//等待任务执行
	isFlushPending = true;
	//将flushJobs放入微任务队列
	currentFlushPromise = resolvedPromise.then(flushJobs);
	}
	}

	function flushJobs(seen) {
	isFlushPending = false; //当前不是等待状态
	isFlushing = true; //当前正在执行任务
	seen = seen \|\| new Map();
	//原文译文:
	//在flush之前对queue排序这样做是为了:
	//.组件更新是重父组件到子组件(因为父组件总是在子组件之前创建
	//所以父组件的render副作用将会有更低的优先级
	//.如果子组件在父组件更新期间并未挂载,那么可以跳过
	queue.sort(comparator);
	//监测当前任务是否已经超过了最大递归层数
	const check = (job) => checkRecursiveUpdates(seen, job);
	try {
	for (flushIndex =; flushIndex < queue.length; flushIndex++) {
	const job = queue[flushIndex];
	if (job && job.active !== false) {
	if (check(job)) {
	continue;
	}
	callWithErrorHandling(job, null,);
	}
	}
	} finally {
	//执行完所有的任务之后,初始化queue
	//调用post任务,这些任务调用完折后
	//可能在执行这些任务的途中还有新的
	//任务加入所以需要继续执行flushJobs
	flushIndex =;
	queue.length =;
	flushPostFlushCbs(seen);
	isFlushing = false;
	currentFlushPromise = null;
	if (queue.length \|\| pendingPostFlushCbs.length) {
	flushJobs(seen);
	}
	}
	}

	function queuePostFlushCb(cb) {
	if (!shared.isArray(cb)) {
	if (
	!activePostFlushCbs \|\|
	!activePostFlushCbs.includes(
	cb,
	cb.allowRecurse ? postFlushIndex + : postFlushIndex
	)
	) {
	pendingPostFlushCbs.push(cb);
	}
	} else {
	pendingPostFlushCbs.push(...cb);
	}
	queueFlush();
	}

	function queueEffectWithSuspense(fn, suspense) {
	//对suspense的处理,暂时不详细解释
	if (suspense && suspense.pendingBranch) {
	if (shared.isArray(fn)) {
	suspense.effects.push(...fn);
	} else {
	suspense.effects.push(fn);
	}
	} else {
	//如果是普通的任务则放入后置队列
	queuePostFlushCb(fn);
	}
	}

	function flushPostFlushCbs(seen) {
	if (pendingPostFlushCbs.length) {
	//克隆等待执行的pendingPost
	const deduped = [...new Set(pendingPostFlushCbs)];
	pendingPostFlushCbs.length =; //设置为0
	//当前函数是后置队列的任务发起的,那么不能
	//直接运行任务,而是将任务放到avtivePostFlushCbs任务之后
	if (activePostFlushCbs) {
	activePostFlushCbs.push(...deduped);
	return;
	}
	activePostFlushCbs = deduped;
	seen = seen \|\| new Map();
	//排序(post依然有优先级)
	activePostFlushCbs.sort((a, b) => getId(a) - getId(b));
	for (
	postFlushIndex =;
	postFlushIndex < activePostFlushCbs.length;
	postFlushIndex++
	) {
	//检测执行深度
	if (checkRecursiveUpdates(seen, activePostFlushCbs[postFlushIndex])) {
	continue;
	}
	//调用这个postJob
	activePostFlushCbs[postFlushIndex]();
	}
	//初始化
	activePostFlushCbs = null;
	postFlushIndex =;
	}
	}

	function flushPreFlushCbs(seen, i = isFlushing ? flushIndex + : 0) {
	seen = seen \|\| new Map();
	for (; i < queue.length; i++) {
	const cb = queue[i];
	if (cb && cb.pre) {
	if (checkRecursiveUpdates(seen, cb)) {
	continue;
	}
	queue.splice(i,);
	i--;
	cb();
	}
	}
	}

	function a(){
	console.log()
	}
	function b(){
	console.log()
	}
	a.pre = true
	queueJobs(a)
	queueJobs(b)
	flushPreFlushCbs()
	//打印: 111

	<script>
	import { nextTick } from 'vue'
	export default {
	data() {
	return { count: } },
	methods: {
	async increment() {
	this.count++ // DOM 还未更新
	//
	console.log(document.getElementById('counter').textContent)
	await nextTick() // DOM 此时已经更新
	console.log(document.getElementById('counter').textContent) //
	}
	}
	}
	</script>
	<template>
	<button id="counter" @click="increment">{{ count }}</button>
	</template>

	function nextTick(fn) {
	const p = currentFlushPromise \|\| resolvedPromise;
	return fn ? p.then(this ? fn.bind(this) : fn) : p;
	}

	<script>
	export default {
	watch{
	a(){},
	b:"meth"//在methods中声明的方法
	c:{
	handler(val,oldVal){},
	deep:true,//开启深度监视
	immediate:true//立即调用handler
	},
	"d.a":function(){}
	}
	}
	</script>

	const callback = ([aOldVal,aVal],[bOldVal,bVal])=>{}
	//监听源监听源发生改变的回调函数选项
	watch(["a","b"], callback, {
	flush: 'post',
	onTrack(e) {
	debugger
	},
	deep:true,
	immediate:true,
	})

	//这一段代码在Vue源码分析(7)中出现过
	//不了解的可以看看上一篇文章
	//对每一个watch选项添加watcher
	if (watchOptions) {
	for (const key in watchOptions) {
	createWatcher(watchOptions[key], ctx, publicThis, key);
	}
	}

	function createWatcher(raw, ctx, publicThis, key) {
	//可以监听深度数据例如a.b.c
	const getter = key.includes(".")
	? createPathGetter(publicThis, key)
	: () => publicThis[key];
	//raw可以是字符串,会读取methods中的方法
	if (shared.isString(raw)) {
	const handler = ctx[raw];
	if (shared.isFunction(handler)) {
	//进行监听
	watch(getter, handler);
	} else {
	warn(`Invalid watch handler specified by key "${raw}"`, handler);
	}
	}
	//如果是函数监听
	else if (shared.isFunction(raw)) {
	watch(getter, raw.bind(publicThis));
	}
	//如果是对象
	else if (shared.isObject(raw)) {
	//数组遍历,获取每一个监听器在执行createWatcher
	if (shared.isArray(raw)) {
	raw.forEach((r) => createWatcher(r, ctx, publicThis, key));
	}
	//对象
	else {
	//handler可能是字符串重ctx上获取
	//也可能是函数
	//获取到handler后调用watch
	const handler = shared.isFunction(raw.handler)
	? raw.handler.bind(publicThis)
	: ctx[raw.handler];
	if (shared.isFunction(handler)) {
	watch(getter, handler, raw);
	} else {
	warn(
	`Invalid watch handler specified by key "${raw.handler}"`,
	handler
	);
	}
	}
	} else {
	warn(`Invalid watch option: "${key}"`, raw);
	}
	}

	function watch(source, cb, options) {
	//cb必须是函数
	if (!shared.isFunction(cb)) {
	console.warn();
	}
	return doWatch(source, cb, options);
	}

	function doWatch(
	source, //getter ()=>[监听的数据]
	cb, //回调函数
	//获取当前watch的选项
	{ immediate, deep, flush, onTrack, onTrigger } = shared.EMPTY_OBJ
	) {
	//immediate和deep属性必须有cb
	if (!cb) {
	if (immediate !== undefined) {
	warn(
	`watch() "immediate" option is only respected when using the ` +
	`watch(source, callback, options?) signature.`
	);
	}
	if (deep !== undefined) {
	warn(
	`watch() "deep" option is only respected when using the ` +
	`watch(source, callback, options?) signature.`
	);
	}
	}
	//省略第二部分代码
	}

	//获取当前实例
	const instance = getCurrentInstance();
	let getter;
	let forceTrigger = false; //强制触发
	let isMultiSource = false; //是否多个数据
	//判断监听的数据是否是ref
	if (reactivity.isRef(source)) {
	getter = () => source.value;
	forceTrigger = reactivity.isShallow(source);
	}
	//判断数据是否是响应式
	else if (reactivity.isReactive(source)) {
	getter = () => source;
	deep = true;
	}
	//判断数据是否是数组
	else if (shared.isArray(source)) {
	isMultiSource = true;
	//source中有一个是响应式的
	//就需要触发
	forceTrigger = source.some(
	(s) => reactivity.isReactive(s) \|\| reactivity.isShallow(s)
	);
	//()=>[proxy,()=>proxy,ref]
	getter = () =>
	source.map((s) => {
	if (reactivity.isRef(s)) {
	return s.value;
	} else if (reactivity.isReactive(s)) {
	//遍历响应式对象s 这个getter会作为ReactiveEffect的
	//第一个参数,在调用run的时候遍历所有的值
	//确保能让每一个变量都能收集到effect
	return traverse(s);
	}
	//调用监听的函数
	else if (shared.isFunction(s)) {
	return callWithErrorHandling(s, instance,);
	} else {
	//提示非法source信息
	warnInvalidSource(s);
	}
	});
	}
	//省略第三部分代码

	//()=>[proxy]传入的是一个函数
	else if (shared.isFunction(source)) {
	if (cb) {
	//让getter为这个函数
	getter = () => callWithErrorHandling(source, instance,);
	} else {
	//如果没有回调函数
	getter = () => {
	if (instance && instance.isUnmounted) {
	return;
	}
	if (cleanup) {
	cleanup();
	}
	return callWithAsyncErrorHandling(source, instance,, [onCleanup]);
	};
	}
	}
	//省略第四部分代码

Vue3源码分析调度器与watch用法原理

目录

本文主要内容

调度器

1.添加任务(queueJobs)

2.二分法找到插入位置(findInsertionIndex)

3.将执行任务的函数推入微任务队列(queueFlush)

4.执行普通任务(flushJobs)

5.添加后置任务(queuePostFlushCb)

6.queuePostRenderEffect

7.执行后置队列任务(flushPostFlushJobs)

8.执行前置任务队列(flushPreFlushCbs)

9.nextTick

调度器总结

watch用法

选项式watch Api的实现

创建watch对象(createWatchr)

选项式watch Api总结

函数式watch的实现(下面统称watch)

1.watch

2.doWatch

watch总结

	watch(async (onCleanup) => {
	const { response, cancel } = doAsyncWork(id.value)
	// `cancel` 会在 `id` 更改时调用
	// 以便取消之前
	// 未完成的请求
	onCleanup(cancel)
	data.value = await response
	})

	//不是以上情况,让getter为空函数
	else {
	getter = shared.NOOP;
	//警告
	warnInvalidSource(source);
	}
	//省略第五部分代码

	const INITIAL_WATCHER_VALUE = {}
	//getter作为参数传入ReactiveEffect
	//调用run的时候会调用getter,确保
	//所有的属性都能够收集到依赖
	if (cb && deep) {
	const baseGetter = getter;
	getter = () => traverse(baseGetter());
	}
	let cleanup;
	//调用effect.stop的时候触发这个函数
	let onCleanup = (fn) => {
	cleanup = effect.onStop = () => {
	callWithErrorHandling(fn, instance,);
	};
	};
	let oldValue = isMultiSource ? [] : INITIAL_WATCHER_VALUE;
	//省略第六部分代码

	//回调函数
	const job = () => {
	if (!effect.active) {
	return;
	}
	//传递了cb函数
	if (cb) {
	//watch([a,b],()=>{})
	//newValue=[a,b]
	const newValue = effect.run();
	//未设置deep属性的
	//旧值和新值要发生改变才会调用cb回调函数
	if (
	deep \|\|
	forceTrigger \|\|
	(isMultiSource
	? newValue.some((v, i) => shared.hasChanged(v, oldValue[i]))
	: shared.hasChanged(newValue, oldValue))
	) {
	//这里的作用上面我们已经讲过了,不在赘述。
	if (cleanup) {
	cleanup();
	}
	callWithAsyncErrorHandling(cb, instance,, [
	newValue,
	oldValue === INITIAL_WATCHER_VALUE ? undefined : oldValue,
	onCleanup,
	]);
	oldValue = newValue;
	}
	} else {
	//没有cb就只调用getter函数(watchEffect)
	effect.run();
	}
	};
	//省略第七部分代码

	//只要有cb则允许递归
	job.allowRecurse = !!cb;
	let scheduler;
	//设置了sync则同步调度,不放入queue进行异步调度(同步)
	if (flush === "sync") {
	scheduler = job;
	}
	//设置了post放到DOM渲染之后执行(异步)
	else if (flush === "post") {
	scheduler = () => queuePostRenderEffect(job, instance && instance.suspense);
	}
	//默认值为pre,放入到queue中执行(异步)
	//带有pre的会在DOM渲染前执行
	else {
	job.pre = true;
	//给当前的job设置优先级
	if (instance) job.id = instance.uid;
	scheduler = () => queueJob(job);
	}
	//省略第八部分代码

	const effect = new reactivity.ReactiveEffect(getter, scheduler);
	//将用户传递的onTrack和onTrigger赋值到effect上
	//便于在track和trigger的时候调用
	effect.onTrack = onTrack;
	effect.onTrigger = onTrigger;
	//省略第九部分代码

	//调用了watch之后
	//需要立刻执行getter,处理不同的flush参数
	if (cb) {
	if (immediate) {
	//有immediate参数立即执行job
	job();
	}
	//否则就只收集依赖调用getter函数
	//并且获取监听的变量
	else {
	oldValue = effect.run();
	}
	}
	//flush为post需要将收集依赖函数getter
	//放到postQueue中
	else if (flush === "post") {
	queuePostRenderEffect(
	effect.run.bind(effect),
	instance && instance.suspense
	);
	}
	//没有设置则收集依赖
	else {
	effect.run();
	}
	//省略第十部分代码

	//watch的停止函数,调用后不再依赖更新
	return () => {
	effect.stop();
	};