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

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

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

//找到插入的位置
//例如[1,2,3,8,9,10,100]
//当前插入的id为20
//插入后应该为[1,2,3,8,9,10,20,100]
//也就是说最终返回的start=6
//插入流程解析:
//1.假设当前执行到第二个任务即flushIndex=2
//那么start = 2;end = 7;middle=4;
//    middleJobId=9;9<20 start=5;
//继续循环:middle=6;middleJobId=100;end=6
//结束循环start = 6;这就是需要插入的位置
function findInsertionIndex(id) {
  let start = flushIndex + 1;
  let end = queue.length;
  while (start < end) {
    // 1000>>>1=>100 8=>4
    // 1100>>>1=>110 12=>6
    // 1010>>>1=>101 10=>5
    // 1001>>>1=>100 9=>4
    //计算出中间值,向下取整
    const middle = (start + end) >>> 1;
    //获取job的id
    const middleJobId = getId(queue[middle]);
    middleJobId < id ? (start = middle + 1) : (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排序这样做是为了:
  //1.组件更新是重父组件到子组件(因为父组件总是在子组件之前创建
  //所以父组件的render副作用将会有更低的优先级
  //2.如果子组件在父组件更新期间并未挂载,那么可以跳过
  queue.sort(comparator);
  //监测当前任务是否已经超过了最大递归层数
  const check = (job) => checkRecursiveUpdates(seen, job);
  try {
    for (flushIndex = 0; flushIndex < queue.length; flushIndex++) {
      const job = queue[flushIndex];
      if (job && job.active !== false) {
        if (check(job)) {
          continue;
        }
        callWithErrorHandling(job, null, 14);
      }
    }
  } finally {
    //执行完所有的任务之后,初始化queue
    //调用post任务,这些任务调用完折后
    //可能在执行这些任务的途中还有新的
    //任务加入所以需要继续执行flushJobs
    flushIndex = 0;
    queue.length = 0;
    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 + 1 : 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; //设置为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 = 0;
      postFlushIndex < activePostFlushCbs.length;
      postFlushIndex++
    ) {
      //检测执行深度
      if (checkRecursiveUpdates(seen, activePostFlushCbs[postFlushIndex])) {
        continue;
      }
      //调用这个postJob
      activePostFlushCbs[postFlushIndex]();
    }
    //初始化
    activePostFlushCbs = null;
    postFlushIndex = 0;
  }
}

function flushPreFlushCbs(seen, i = isFlushing ? flushIndex + 1 : 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, 1);
      i--;
      cb();
    }
  }
}

job.pre = true

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

<script> 
import { nextTick } from 'vue' 
export default {  
  data() {   
    return { count: 0 } }, 
    methods: { 
      async increment() { 
        this.count++ // DOM 还未更新
        // 0
        console.log(document.getElementById('counter').textContent)
        await nextTick() // DOM 此时已经更新 
        console.log(document.getElementById('counter').textContent) // 1 
        } 
     } 
 } 
 </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;
}

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

<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,
})

//这一段代码在Vue3源码分析(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, 2);
      } else {
        //提示非法source信息
        warnInvalidSource(s);
      }
    });
}
//省略第三部分代码

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

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, 4);
  };
};
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, 3, [
        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();
};