教你如何实现react Scheduler（二）

标签：PRIORITY 优先级 常量 队列 react 如何 任务 Scheduler 执行

在上一篇文章中 “反应调度程序（1）” 中，我们使用 MessageChannel 来实现一个简单的任务调度功能。但是这个功能目前还是比较简单的，现在我们来改进一下，给任务增加优先级和过期时间。

定义任务优先级：

 常量 NoPriority = 0; // 没有优先级  
 常量立即优先级 = 1; // 最高优先级  
 常量用户阻塞优先级 = 2;  
 常量 NormalPriority = 3;  
 常量低优先级 = 4;  
 常量空闲优先级 = 5; // 最低优先级  
 复制代码

首先要明确一点，在任务队列taskQueue中，任务的排序并不是直接根据这个优先级，而是根据过期时间。较早到期的任务将较早地放入队列中。所以我们还需要定义任务延迟时间：

 常量 maxSigned31BitInt = 1073741823; // 最大的 31 位整数  
 常量 IMMEDIATE_PRIORITY_TIMEOUT = - 1; // 立即过期，对应最高优先级的任务  
 常量 USER_BLOCKING_PRIORITY_TIMEOUT = 250；  
 常量 NORMAL_PRIORITY_TIMEOUT = 5000;  
 常量 LOW_PRIORITY_TIMEOUT = 10000;  
 常量 IDLE_PRIORITY_TIMEOUT = maxSigned31BitInt; //永不过期  
 复制代码

任务开始时间+延迟时间=任务到期时间。在执行任务时，我们只会在任务到期时执行该任务，否则我们将交出主线程的控制权。

既然我们已经有了任务优先级，那我们就不能直接按照任务优先级来排序吗？为什么我们需要设置过期时间？假设这样一个场景：我们将一个优先级为 NormalPriority（切片为 A1、A2、A3）的任务 A 添加到队列中。在调度A1的时候，我们在队列中加入了一个优先级为USER_BLOCKING_PRIORITY_TIMEOUT的任务B，然后在任务B结束后，继续加入任务B，这样队列就被切掉了。如果只根据优先级决定先执行哪个任务，那么任务A永远不会执行。但是现在我们使用过期时间来按过期时间排序。一开始有任务B来切队。因为你的任务B的过期时间只有250，expirationTime比较小，ok没问题，可以切队列先执行。随着你切队列的次数越来越多，你后面插入的任务的expirationTime肯定会比任务A的要大。这个时候，你应该等待任务A先执行。这就是 expireTime 所做的：避免低优先级的任务永远不会被执行。

现在任务不仅仅是一个简单的函数，我们使用函数 createTask 来创建任务：

 让 taskIdCounter = 1; // 全局变量，自增任务id  
 函数创建任务（优先级，回调）{  
 const currentTime = 性能。现在（）;  
 常量开始时间 = 当前时间；  
 让超时 = 0;  
 开关（优先级）{  
 案例立即优先级：  
 超时 = IMMEDIATE_PRIORITY_TIMEOUT；  
 休息;  
 案例用户阻塞优先级：  
 超时 = USER_BLOCKING_PRIORITY_TIMEOUT；  
 休息;  
 情况空闲优先级：  
 超时 = IDLE_PRIORITY_TIMEOUT；  
 休息;  
 案例低优先级：  
 超时 = LOW_PRIORITY_TIMEOUT；  
 休息;  
 案例 NormalPriority：  
 默认：  
 超时 = NORMAL_PRIORITY_TIMEOUT；  
 休息;  
 }  
 常量过期时间 = 开始时间 + 超时； // 计算不同优先级任务对应的过期时间  
 常量新任务 = {  
 id: taskIdCounter++,  
 callback, // 待执行的任务  
 优先级，  
 开始时间，  
 过期时间，  
 sortIndex: expirationTime, // 使用 expireTime 排序  
 }  
 返回新任务；  
 }  
 复制代码

任务已经创建，我们需要有一个优先级队列来存储任务。优先级任务队列是最小堆数据结构。这里简单介绍一下最小堆，不熟悉的读者可以自行理解。

堆是一棵完全二叉树，其中每个节点都小于或等于其子节点。数组可用于在 js 中实现最小堆。对于数组索引为index的节点，其父节点索引为 Math.floor(index - 1) / 2 , 左右子节点的索引分别为 2 * 索引 + 1 和 2 * 索引 + 2 .最小堆的顶部元素是最小值，因此获取最小值的时间复杂度为 O(1)。对应我们的任务队列，堆顶是过期时间最短的任务（需要最早执行）。

最小堆的js实现：

 类 SchedulerMinHeap {  
 构造函数（）{  
 这个。堆 = [];  
 }  
 推（节点）{  
 这个。堆。推（节点）；  
 这个。 siftUp(node, this.heap.length - 1);  
 }  
 窥视（）{  
 返回堆。长度 ？堆[0]：空；  
 }  
 流行音乐（）{  
 if(this.heap.length === 0) 返回空值；  
 if(this.heap.length === 1) 返回这个。堆。流行音乐（）;  
 const first = heap[0];  
 常量最后 = 这个。堆。流行音乐（）;  
 堆[0] = 最后一个； // 将最后一个节点放在堆顶  
 这个。 siftDown(堆[0], 0); // 开始向下移动  
 先返回； // 返回堆的顶部节点  
 }  
 // 向上移动节点直到到达堆顶或父节点小于自身  
 筛选（节点，idx）{  
 让索引 = idx;  
 而（索引> 0）{  
 常量父索引 = 数学。地板（索引 - 1）；  
 常量父 = 这个。堆[父索引]；  
 如果（这个。比较（节点，父）<0）{  
 // 小于父节点，交换位置  
 头[父索引] = 节点；  
 头部 [索引] = 父级；  
 索引 = 父索引；  
 } 别的 {  
 返回;  
 }  
 }  
 }  
 // 向下移动节点，直到到达堆底部或子节点大于自身。  
 // 调用场景：删除堆顶节点后，将堆底元素放在堆顶，然后向下移动  
 siftDown（节点，idx）{  
 让索引 = i;  
 常量长度 = 这个。堆。长度;  
 常量一半 = 数学。地板（长度/2）； // 在react源码中使用移位操作：index >>> 1  
 而（索引<一半）{  
 // 还没有到达二叉树的底部  
 const leftIndex = (index + 1) * 2 - 1;  
 const rightIndex = leftIndex + 1;  
 // 左子节点必须存在。如果不存在，说明已经到了二叉树的底部，不会进入这个循环。  
 常量左 = 这个。堆[leftIndex];  
 // 右子节点不一定存在  
 const 对 = 这个。堆[rightIndex];  
 如果（这个。比较（左，节点）<0）{  
 // 左子节点小于当前节点。需要下移，交换左子节点还是右子节点？  
 if(rightIndex < length && compare(right, left) < 0) {  
 // 有右子节点且较小，交换右子节点  
 这个。堆[索引] = 对；  
 这个。堆[rightIndex] = 节点；  
 索引 = 右索引；  
 } 别的{  
 // 左孩子变小，交换左孩子  
 这个。堆[索引] = 左；  
 这个。堆[leftIndex] = 节点；  
 索引 = 左索引；  
 }  
 } else if(rightIndex < length && compare(right, node) < 0) {  
 // 左子节点比较大，看右子节点情况：右子节点存在且较小，交换  
 这个。堆[索引] = 对；  
 这个。堆[rightIndex] = 节点；  
 索引 = 右索引；  
 } 别的 {  
 // 左右子节点都比节点大  
 返回;  
 }  
 }  
 }  
 // a小于b，则返回负数；  
 比较（一，乙）{  
 // 先用sortIndex比较，再用id比较  
 常量差异 = a。排序索引 - b。排序索引；  
 返回差异！== 0？差异：一个。身份证 - b。 ID;  
 }  
 }  
 复制代码

实例化一个全局任务队列：

 const taskQueue = new SchedulerMinHeap();  
 复制代码

任务和任务队列已经实现。接下来要做的是创建一个任务，将其推送到任务队列中，然后postMessage开始任务调度。

 函数 scheduleCallback（优先级，回调）{  
 const task = createTask(priorityLevel, callback); // 创建任务  
 任务队列。推（任务）； //加入任务队列  
 请求主机回调（）； // 开始调度任务  
 }  
 复制代码

接下来是requestHostCallback的实现。需要注意的是，scheduleCallback 可能会被多次调用，不可能每次调用都postMessage。所以我们将使用一个全局变量来控制它：

 常量频道 = 新的 MessageChannel();  
 常量端口 2 = 通道。端口2；  
 常量端口 1 = 通道。端口1；  
 端口 1。 onmessage = performWorkUntilDeadline;  
  
 让 isMessageLoopRunning = false; // 主动调度任务时，如果该值为true，则不能postMessage  
 函数请求主机回调（）{  
 如果（！isMessageLoopRunning）{  
 端口 2。 postMessage(空);  
 }  
 }  
  
 功能 performWorkUntilDeadline() {  
 让 hasMoreWork = true;  
 const currentTime = 性能。现在（）;  
 开始时间 = 当前时间； //更新任务开始时间  
 尝试 {  
 hasMoreWork = flushWork(currentTime);  
 } 最后 {  
 如果（有更多工作）{  
 端口 2。 postMessage(空);  
 } 别的 {  
 // 没有任务了，等待下一次创建新任务  
 isMessageLoopRunning = 假；  
 }  
 }  
 }  
 复制代码

可以看到，从任务队列中取出并执行的任务都是在flushWork中完成的，它返回任务队列中是否还有任务需要处理。

 let startTime = - 1; 任务开始时间  
 常量 frameYieldMs = 5; // 任务的连续执行时间不能超过5ms  
 让 currentTask = null; // 用于保存当前任务  
  
 函数flushWork（初始时间）{  
 尝试 {  
 返回工作循环（初始时间）；  
 } 最后 {  
 当前任务=空；  
 }  
 }  
  
 函数工作循环（初始时间）{  
 让当前时间 = 初始时间；  
 // 取出队列中的第一个任务，注意这里用的是peek，不是pop，任务还在队列中  
 当前任务 = 任务队列。窥视（）;  
 而（当前任务！== null）{  
 if(currentTask.expirationTime > currentTime && shouldYieldToHost()) {  
 // 过期时间未到，调度已超过5ms  
 休息;  
 }  
 // 可以安排任务  
 常量回调 = 当前任务。打回来;  
 if (typeof callback === 'function') {  
 当前任务。回调=空；  
 常量 continuationCallback = 回调（）； // 做真正的任务  
 if ( typeof continuationCallback === 'function') {  
 // 注意这很关键：如果我们的任务返回一个函数，这意味着该任务是一个分片任务  
 // 第一个shard任务执行后，不会再接下一个任务，而是重新分配task.callback，  
 // 然后再经过while循环，执行下一个分片任务。每个分片任务执行前，都要判断是否超过5ms。  
 // 如果超过了，就会中断，等待下一个事件循环被取出并再次执行。  
 当前任务。回调 = 延续回调；  
 } 别的 {  
 // 任务不分片，执行后可删除  
 if(currentTask === taskQueue.peek()) {  
 任务队列。流行音乐（）;  
 }  
 }  
 } 别的 {  
 // 对于任务分片的情况，所有分片的任务都已经执行完毕，  
 // 表示当前任务已经完成，可以删除  
 任务队列。流行音乐（）;  
 }  
 currentTask = peek(taskQueue); // 取下一个任务执行  
 }  
 如果（当前任务！== null）{  
 // 在下一个事件循环中还有任务需要处理  
 返回真；  
 }  
 }  
  
 函数应该YieldToHost() {  
 // 任务是否应该被挂起  
 const currentTime = 性能。现在（）;  
 如果（当前时间 - 开始时间 < frameYieldMs）{  
 返回假；  
 }  
 返回真；  
 }  
  
 复制代码

如果你理解了workLoop的逻辑，你会发现我们实际上已经实现了高优先级任务的队列切割功能。需要注意的是，定时任务回调一定要有好的设计，要么是短时间，要么是一个shard任务，每个shard耗时很短。分片任务设计的关键是在每次执行前使用 shouldYieldToHost 函数判断任务调度是否已经达到 5ms。如果达到 5ms，则返回一个函数，调度器将在下一个事件循环中继续执行它。同时，这个任务还必须记住执行的进度，否则每次从头执行都会陷入死循环。

以下是高优先级任务如何切队列的解释：

如果当前正在调度一个分片任务A（分为A1、A2、A3），​​当A1正在执行时，调用scheduleCallback调度一个高优先级任务B，该任务将被推入任务队列（最小堆，根据优先级排序）。此时需要等待A1执行完毕，判断是否超时。如果没有超时，继续执行A2。 A2执行完后发现已经超时，跳出循环。当进入下一个事件循环并重新进入workLoop函数时，任务再次从任务队列中取出。此时取出任务B，将任务B切入队列。 B执行完后，继续取出A执行。再次强调，A 任务必须有正确的分片设计，取出时会从 A3 开始执行。

经过分析，我们知道任务插入不是随机的，最早只能在下一个事件循环之后执行。

至此，react scheduler的核心功能已经实现。其实在react源码中，不仅有一个任务队列taskQueue，还有一个timerQueue，用来处理一些延迟的任务。在某些场景下，我们在创建任务时，并不想立即将其添加到taskQueue中，而是延迟一段时间再添加到taskQueue中，然后再将其添加到timerQueue中。实现这个会使得调度器的逻辑很复杂，影响我们对最核心原理的把握，所以暂时不实现。

版权声明：本文为博主原创文章，遵循CC 4.0 BY-SA版权协议。转载请附上原文出处链接和本声明。

这篇文章的链接： https://homecpp.art/0522/10141/1134

版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明

本文链接：https://www.qanswer.top/38696/47482212

标签：PRIORITY,优先级,常量,队列,react,如何,任务,Scheduler,执行
From： https://www.cnblogs.com/amboke/p/16718853.html