剖析React系列十二-调度器的实现

本系列是讲述从0开始实现一个react18的基本版本。通过实现一个基本版本，让大家深入了解React内部机制。 由于React源码通过Mono-repo 管理仓库，我们也是用pnpm提供的workspaces来管理我们的代码仓库，打包我们使用rollup进行打包。

仓库地址

系列文章：

1. React实现系列一 - jsx
2. 剖析React系列二-reconciler
3. 剖析React系列三-打标记
4. 剖析React系列四-commit
5. 剖析React系列五-update流程
6. 剖析React系列六-dispatch update流程
7. 剖析React系列七-事件系统
8. 剖析React系列八-同级节点diff
9. 剖析React系列九-Fragment的部分实现逻辑
10. 剖析React系列十- 调度<合并更新、优先级>
11. 剖析React系列十一- useEffect的实现原理

我们从之前的章节中晓得，React每次更新都是从根节点开始，相比于Vue模板组件维度的更新，是有一点庞大的，当遇到一个复杂的应用如果没有合适的手段，就是长时间占据js主线程，而浏览器在执行js的时候，会阻塞浏览器渲染和绘制，所以会出现卡顿。

react通过调度器和浏览器之间进行”君子交易”，大体是，浏览器一帧执行大概16.6ms, 调度器使用其中的5ms。

本节主要是通过以下几个方面讲解：

1. scheduler调度器的实现
2. react异步调度原理
3. 时间分片
4. react的调度执行流程

scheduler调度器的实现

react将调度器scheduler单独抽出一个包，和主流程分开，我们来看看scheduler中的逻辑。

一些核心方法

*scheduleCallback: 以某一优先级调度callback

• shouldYield: 提示Time Slice 时间是否用尽
• cancelCallback: 移除task

优先级和任务

在scheduler预置了5个优先级，用于代表事情的紧急度。每一种的优先级对应任务过期时间也不一样。

• ImmediatePriority = 1: 最高优先级，同步执行
• UserBlockingPriority = 2
• NormalPriority = 3
• LowPriority
• IdlePriority

不同的优先级，对应不同的超时时间timeout，这个在下面会用到。

• IMMEDIATE_PRIORITY_TIMEOUT = -1
• USER_BLOCKING_PRIORITY_TIMEOUT = 250
• NORMAL_PRIORITY_TIMEOUT = 5000
• LOW_PRIORITY_TIMEOUT = 10000

scheduler对外暴露的函数scheduleCallback,它接收一个优先级和一个回调函数Fn, 根据优先级调度Fn的执行。

// 以NormalPriority优先级调度回调函数fn
scheduleCallback(NormalPriority, Fn)

我们来看看源码中scheduleCallback具体是逻辑，主要是用来创建task。

scheduleCallback逻辑

scheduleCallback主要是接受优先级和回调函数，然后生成一个task用于内部的调用。它包含三个参数

1. priorityLevel 是调度任务的优先级；
2. callback 是需要执行的更新任务；
3. optoins 里面可以通过指定 delay 延迟执行任务 或 timeout 定义任务的超时时间。

任务的开始时间和过期时间

1. 任务的开始时间startTime: 决定了之后放入到taskQueue还是timerQueue
2. 任务的过期时间expirationTime: 决定了任务在执行的队列中的优先级，值越小优先级越高。高优先级的task会优先执行（task.callback）

// main.js 执行的入口文件，传递不同的优先级
export function unstable_scheduleCallback(priorityLevel, callback, options) {
  // 获取当前的时间（ms)
  var currentTime = unstable_now(); // performance.now();
  var startTime;

  // 这里主要是查看是否设置了delay, 默认的开始时间是当前时间
  if (typeof options === "object" && options !== null) {
    var delay = options.delay;

    if (typeof delay === "number" && delay > 0) {
      startTime = currentTime + delay;
    } else {
      startTime = currentTime;
    }
  } else {
    startTime = currentTime;
  }

  var timeout;

  // 不同优先级设置不同的超时时间
  switch (priorityLevel) {
    case ImmediatePriority:
      timeout = IMMEDIATE_PRIORITY_TIMEOUT;
      break;

    case UserBlockingPriority:
      timeout = USER_BLOCKING_PRIORITY_TIMEOUT;
      break;

    case IdlePriority:
      timeout = IDLE_PRIORITY_TIMEOUT;
      break;

    case LowPriority:
      timeout = LOW_PRIORITY_TIMEOUT;
      break;

    case NormalPriority:
    default:
      timeout = NORMAL_PRIORITY_TIMEOUT;
      break;
  }

  /* 计算任务过期时间：超时时间  = 开始时间（现在时间） + 任务超时的时间（上述设置那五个等级）     */
  var expirationTime = startTime + timeout;
}

创建任务task的数据结构

一个任务包含以下字段：

function unstable_scheduleCallback(priorityLevel, callback, options) {
  //...

  var newTask = {
    id: taskIdCounter++, // 任务id
    callback,  // 执行回调
    priorityLevel, // 任务优先级
    startTime, // 任务开始时间
    expirationTime, // 任务结束时间
    sortIndex: -1, // 用于小顶堆的重建排序
  };

 // ...
}

加入任务队列taskQueue、timeQueue

从scheduleCallback中我们看到了如下的2种执行情况：

/* 把任务放在timerQueue中 */ 
push(timerQueue, newTask);
/* 将任务放入taskQueue中 */
push(taskQueue, newTask);

那么它们之间有什么区别呢？ 在scheduleCallback中，根据任务如果标识了delay，会将任务放入timerQueue中，否则就放入taskQueue中。

每一个任务，可以理解为触发了一次React更新

大体的转换如上图所示：

1. timeQueue中的task以currentTime + delay排序
2. taskQueue中的task以expirationTime为排序依据
3. 当timerQueue中第一个task延迟的时间到期后，执行advaceTimers将“到期的task"从timerQueue中移动到taskQueue中。

export function unstable_scheduleCallback(priorityLevel, callback, options) {
  // ....
  if (startTime > currentTime) {
    /* 通过开始时间排序 */
    newTask.sortIndex = startTime;
    /* 把任务放在timerQueue中 */
    push(timerQueue, newTask);

    if (peek(taskQueue) === null && newTask === peek(timerQueue)) {
      // All tasks are delayed, and this is the task with the earliest delay.
      if (isHostTimeoutScheduled) {
        // Cancel an existing timeout.
        cancelHostTimeout();
      } else {
        isHostTimeoutScheduled = true;
      } // Schedule a timeout.

      /*  执行setTimeout ， */
      requestHostTimeout(handleTimeout, startTime - currentTime);
    }
  } else {
    newTask.sortIndex = expirationTime;
    push(taskQueue, newTask);
    /*没有处于调度中的任务， 然后向浏览器请求一帧，浏览器空闲执行 flushWork */
    if (!isHostCallbackScheduled && !isPerformingWork) {
      isHostCallbackScheduled = true;
      requestHostCallback(flushWork);
    }
  }

  return newTask;
}

小顶堆

在scheduler中是通过小顶堆的形式来存放task。 它的特点：任意节点都小于等于其左右节点值。并且是一个完全二叉树，这样就可以很方便的通过数组保存。

主要包含三个方法：

• push: 向堆中推入数据
• pop：从堆顶取出数据
• peek：获取”排序依据最小的值“对应的节点

宏任务调度requestHostCallback

当scheduleCallback处理完成taskQueue之后，通过 requestHostCallback(flushWork)时间切片，异步宏任务的调度flushWork

/* 向浏览器请求执行更新任务 */
function requestHostCallback(callback) {
  scheduledHostCallback = callback; // flushWork

  if (!isMessageLoopRunning) {
    isMessageLoopRunning = true;
    schedulePerformWorkUntilDeadline();
  }
}

其实requestHostCallback中只是把flushWork赋值给scheduledHostCallback, 等待之后调用。

然后执行schedulePerformWorkUntilDeadline, 这个方法就是一个宏任务，在不同的环境使用不同的方法。

通过执行宏任务调用performWorkUntilDeadline，它里面就会去执行真正的flushWork

var performWorkUntilDeadline = function () {
  if (scheduledHostCallback !== null) {
    // flushWork
    var currentTime = unstable_now(); // Keep track of the start time so we can measure how long the main thread
    // has been blocked.

    startTime = currentTime;
    var hasTimeRemaining = true; 
    var hasMoreWork = true;

    try {
      hasMoreWork = scheduledHostCallback(hasTimeRemaining, currentTime); // flushWork
    } finally {
      if (hasMoreWork) {
        // If there's more work, schedule the next message event at the end
        // of the preceding one.
        schedulePerformWorkUntilDeadline();
      } else {
        isMessageLoopRunning = false;
        scheduledHostCallback = null;
      }
    }
  }
  // ...
};

这里需要注意2点：

1. 之后执行flushWork的时候，传入的(true, 当前时间)
2. 如果flushWork的返回值为true的话，代表还有任务未执行，就需要再次执行schedulePerformWorkUntilDeadline, 调用宏任务重新开始调度。

flushWork 执行 taskQueue

flushWork的核心就是通过workLoop同步循环的执行taskQueue中的task.callback.

function flushWork(hasTimeRemaining, initialTime) {
  isHostCallbackScheduled = false;

  /* 因为有taskQueue任务来了，如果有延时任务，那么先暂定延时任务*/
  if (isHostTimeoutScheduled) {
    isHostTimeoutScheduled = false;
    cancelHostTimeout();
  }

  isPerformingWork = true;
  // 将正在执行的优先级修改成旧的优先级
  var previousPriorityLevel = currentPriorityLevel;
  try {
      /* 执行 workLoop 里面会真正调度我们的事件  */
      return workLoop(hasTimeRemaining, initialTime); // true  currentTime
  } finally {
    currentTask = null;
    // 执行完成后回复优先级
    currentPriorityLevel = previousPriorityLevel;
    isPerformingWork = false;
  }
}

workLoop同步调度任务

workLoop接受2个参数，第一个参数一直是true，我们可以删掉它的部分。第二个参数是当前的时间。

function workLoop(hasTimeRemaining, initialTime) {
  var currentTime = initialTime;
  // 查看延时队列中是否有过期的任务，将其加入到taskQueue中
  advanceTimers(currentTime);
  /* 获取任务列表中的第一个 */
  currentTask = peek(taskQueue);

  /* workLoop 会依次更新过期任务队列中的任务 */
  while (currentTask !== null) {
    // 如果当前任务未到期，但是时间切片用完了，中断循环，将执行权交给浏览器，等待下一次执行。
    if (
      currentTask.expirationTime > currentTime && shouldYieldToHost()
    ) {
      break;
    }

    /* 真正的更新函数 callback */
    var callback = currentTask.callback;
    if (typeof callback === "function") {
      currentTask.callback = null;
      currentPriorityLevel = currentTask.priorityLevel;
      // 是否超时
      var didUserCallbackTimeout = currentTask.expirationTime <= currentTime;

      /* 执行callback（传入是否超时） */
      var continuationCallback = callback(didUserCallbackTimeout);
      currentTime = unstable_now();
      // 如果当让出主线程的时候任务还没有执行完成，就返回函数，等待下次调用
      if (typeof continuationCallback === "function") {
        currentTask.callback = continuationCallback;
      } else {
        // 如果任务完成，就返回null
        if (currentTask === peek(taskQueue)) {
          pop(taskQueue);
        }
      }
      // 再次梳理timerQueue
      advanceTimers(currentTime);
    } else {
      // 如果当前任务不存在callback执行，就删除任务
      pop(taskQueue);
    }
    
    // 取下一个最高优先级任务
    currentTask = peek(taskQueue);
  } // Return whether there's additional work

  // 返回taskQueue队列的执行状态：是否有剩余任务工作
  if (currentTask !== null) {
    return true; // 对应hasMoreWork
  } else {
    // 如果taskQueue中执行完了，就开始通过setTimeout延时执行timerQueue的第一个。
    var firstTimer = peek(timerQueue);
    if (firstTimer !== null) {
      requestHostTimeout(handleTimeout, firstTimer.startTime - currentTime);
    }
    return false;
  }
}

这里的流程主要如下：

1. workloop通过while不断的同步从taskQueue中取出第一个要执行的任务。
2. 当时间切片用完shouldYieldToHost() === true的时候，中断循环，让出执行权给浏览器。分2种情况：
   • 当前任务还没有执行完： 需要保存当前的任务进度。等待下一次宏任务调度

     continuationCallback=callback(didUserCallbackTimeout);
     currentTask.callback = continuationCallback`
     

   • 当前任务执行完，但还有其他任务：workLoop返回true执行完毕。flushWork也是返回true。所以会再次执行schedulePerformWorkUntilDeadline(); 重新执行performWorkUntilDeadline调度。

        try {
          hasMoreWork = scheduledHostCallback(hasTimeRemaining, currentTime); // flushWork返回值
        } finally {
          if (hasMoreWork) {
            schedulePerformWorkUntilDeadline();
          }
        }
     

3. 执行 task.callback，梳理 timerQueue；
4. 如果当前任务已经过期，会立刻执行完成，并不会让出执行权。直到过期任务执行完成。
5. 如果taskQueue中执行完成，就延时开始执行timerQueue中的延时任务。

workLoop 每次执行完一个taskQueue task 后，会调用 advanceTimers 方法来看 timerQueue 中是否有任务需要移交到 taskQueue 队列中执行，具体实现如下

function advanceTimers(currentTime) {
  // 获取第一个timer task
  var timer = peek(timerQueue);

  while (timer !== null) {
    if (timer.callback === null) {
      // Timer was cancelled.
      pop(timerQueue);
    } else if (timer.startTime <= currentTime) {
      /* 如果任务已经过期，那么将 timerQueue 中的过期任务，放入taskQueue */
      pop(timerQueue);
      // 在加入 tastQueue 前，将任务的过期时间赋值给 sortIndex，用于任务时间的排序
      timer.sortIndex = timer.expirationTime;
      push(taskQueue, timer);
    } else {
      // Remaining timers are pending.
      return;
    }
    timer = peek(timerQueue);
  }
}

requestHostTimeout处理timerQueue

从workLoop中，我们得知，如果当taskQueue中不存在任务后。会执行如下代码：

  requestHostTimeout(handleTimeout, firstTimer.startTime - currentTime);

requestHostTimeout相当于是setTimeout。所以就等于延时去执行handleTimeout。

function handleTimeout(currentTime) {
  isHostTimeoutScheduled = false;
  /* 将 timeQueue 中过期的任务，放在 taskQueue 中 。 */
  advanceTimers(currentTime);

  /* 如果没有处于调度中 */
  if (!isHostCallbackScheduled) {
    // 是否有调度任务在执行中
    /* advanceTimers 梳理后如果有新任务 */
    if (peek(taskQueue) !== null) {
      isHostCallbackScheduled = true;

      /* 开启调度任务 */
      requestHostCallback(flushWork);
    } else {
      var firstTimer = peek(timerQueue);
      // 如果 taskQueue 仍然为空，就开始递归的调用该方法
      if (firstTimer !== null) {
        requestHostTimeout(handleTimeout, firstTimer.startTime - currentTime);
      }
    }
  }
}

所以handleTimeout会不断的梳理延时任务，当任务被移交到taskQueue后，调用requestHostCallback开始新的调度任务。

例子

下面我们通过一个实际例子来讲解scheduler的运用。

我们有4个优先级的按钮，当低优先级的任务正在执行的时候，点击高级优先级的按钮的话，低优先级的任务会中断。然后执行高优先级任务。

const workList= []; // 代表我们执行任务队列

// 代表这一次的更新有100个子组件的fiberNode调和
workList.unshift({
    count: 100,
    priority: priority,
});

整体效果链接

初始化创建4个优先级按钮

[LowPriority, NormalPriority, UserBlockingPriority, ImmediatePriority].forEach(
  (priority) => {
    const btn = document.createElement("button");
    root?.appendChild(btn);
    btn.innerText = [
      "",
      `ImmediatePriority-${priority}`,
      `UserBlockingPriority-${priority}`,
      `NormalPriority-${priority}`,
      `LowPriority-${priority}`,
    ][priority];
    btn.onclick = () => {
      console.log("-hcc---priority--priority", priority);
      workList.unshift({
        count: 100,
        priority: priority,
      });
      schedule();
    };
  }
);

每一次点击按钮的时候，向任务队列中塞入任务，然后调用schedule函数开始调度。我们来看看schedule中的逻辑。

schedule执行

它主要是分为几点：

1. 类比react更新，我们需要获取到最高优先级的任务执行。
2. 如果执行了调度，但是当前没有任务执行了，就需要正在执行的调度。
3. 如果前后2次优先级相同的话，直接退出。
4. 如果前后2次优先级不同， 高优先级的打断了低优先级的任务，就取消之前正在调度的任务，重新开始新一轮的调度。

function schedule() {
  const cbNode = getFirstCallbackNode();
  // 获取优先级最高的work
  const curWork = workList.sort((w1, w2) => w1.priority - w2.priority)[0];
  // console.log("-hcc--curWork--", curWork, prevPriority);
  //  策略逻辑
  if (!curWork) {
    curCallback = null;
    cbNode && cancelCallback(cbNode);
    return;
  }

  const { priority: currentPriority } = curWork;
  if (currentPriority === prevPriority) {
    return;
  }
  // 更高优先级的work <取消之前的调度>
  cbNode && cancelCallback(cbNode);
  curCallback = scheduleCallback(currentPriority, perform.bind(null, curWork));
}

这里用到scheduler提供的几个方法。

• getFirstCallbackNode方法: 源码中就是查看taskQueue的头部第一个任务。
• cancelCallback方法: 将传入的task的callback赋值为null
• scheduleCallback方法: 传入优先级和任务，放入taskQueue中，开启调度。

这里我们传入的callback为perform.bind(null, curWork)，我们来看看perform执行的内容。

perform的内容

从上面的scheduleCallback讲解中，我们晓得会通过一个宏任务去执行taskQueue。而perform保存在taskQueue中。

function perform(work , didTimeout ) {
  /**
   * 中断情况
   * 1. work.priority <高优先级的优先执行>
   * 2. 饥饿问题<一个work一直得不到执行，优先级就会越来越高> didTimeout
   * 3. 时间切片 <当前执行的时间是否已经完成shouldYield>
   */
  const needSync = work.priority === ImmediatePriority || didTimeout;
  let whileState = performance.now()
  while ((needSync || !shouldYield()) && work.count) {
    console.log('--shouldYield---', shouldYield());
    work.count--;
    insertSpan(work.priority + "");
  }
  console.log('----while---end----', performance.now() - whileState);
  // 中断执行 || 执行完
  prevPriority = work.priority;
  if (!work.count) {
    const workIndex = workList.indexOf(work);
    workList.splice(workIndex, 1);
    prevPriority = IdlePriority;
  }
  const prevCallback = curCallback;
  schedule();
  const newCallback = curCallback;
  if (newCallback && prevCallback === newCallback) {
    return perform.bind(null, work);
  }
}

perform内部主要是分为几个步骤：

1. 判断任务的优先级或者任务是否过期，是否是同步执行。如果是同步执行，就通过while循环直接执行完成。
2. 如果不是同步优先级，如果当前的时间已经执行完，也中断执行。
3. 如果任务没有执行完被中断了，那么再次执行调度schedule函数。这里有一个优化，如果是同一个优先级中断，前后curCallback相同，只需要返回一个函数，scheduler就会记录当前任务为这个函数继续调度。

      /* 执行更新 */
      var continuationCallback = callback(didUserCallbackTimeout); // 传进来的perform的返回值
      currentTime = unstable_now();
      if (typeof continuationCallback === "function") {
        currentTask.callback = continuationCallback;
      }

下一节我们结合react源码，通过几个例子来分析调度器是如何和react结合的。

---

参考文献：

# React Scheduler - 优先级调度