跳到正文

手写练习/异步控制

GitHub

手写调度器 scheduler

困难异步闭包并发队列

代码实现

JavaScript64 行
// 所有任务先统一入队，再由 run() 调度；闭包保存队列和并发计数。
function createScheduler(max) {
  const queue = []
  let running = 0

  function addTo(task) {
    return new Promise((resolve, reject) => {
      queue.push({ task, resolve, reject })
      run()
    })
  }

  function run() {
    while (running < max && queue.length > 0) {
      const { task, resolve, reject } = queue.shift()
      running++

      // 在 Promise 链中调用 task，统一处理返回值和错误。
      Promise.resolve()
        .then(task)
        .then(resolve, reject)
        .finally(() => {
          running--
          run()
        })
    }
  }

  return addTo
}

// 只有槽位已满才入队；队列保存包装好的 runTask。
function createScheduler2(max) {
  const queue = []
  let running = 0

  function addTo(task) {
    return new Promise((resolve, reject) => {
      const runTask = () => {
        running++

        Promise.resolve()
          .then(task)
          .then(resolve, reject)
          .finally(() => {
            running--

            if (queue.length > 0) {
              const nextTask = queue.shift()
              nextTask()
            }
          })
      }

      if (running < max) {
        runTask()
      } else {
        queue.push(runTask)
      }
    })
  }

  return addTo
}

笔记记录

手写调度器 scheduler

要解决什么问题

任务可以不断提交，但同时占用的执行槽位不能超过 max。超出的任务排队，某个任务完成或失败后，后面的任务再补上。

下面保留两种闭包写法，都用 queue 和 running 控制并发，每次添加任务都返回它自己的 Promise：

  • createScheduler：所有任务先入队，再由同一个 run() 调度。
  • createScheduler2：有空位就安排执行，只有槽位已满才把包装好的函数入队。

调用方式与执行时机

const addTask = createScheduler(2)
const delay = (value, ms) => () =>
  new Promise(resolve => setTimeout(() => resolve(value), ms))
 
addTask(delay('A', 100)).then(console.log).catch(console.error)
addTask(delay('B', 30)).then(console.log).catch(console.error)
addTask(delay('C', 10)).then(console.log).catch(console.error)
 
// A、B 先获得槽位，C 排队。
// B 完成后 C 才启动，正常计时下输出顺序为 B、C、A。

createScheduler(2) 只创建调度器并返回 addTo，此时没有任务执行。调用 addTask(task) 才会入队并触发调度。

run() 同步取出任务并增加 running，先占住槽位；真正的 task() 在 .then(task) 的微任务中调用。因此 running 包括已经获得槽位、即将开始执行的任务。

主流程

先看第一种“统一入队”的写法：

步骤代码作用
提交queue.push({ task, resolve, reject })把任务和它自己的 Promise 完成函数绑在一起
取任务queue.shift()按提交顺序取出等待任务
占槽位running++在任务真正开始前更新并发计数
执行.then(task)启动任务并等待返回结果
通知调用方.then(resolve, reject)完成或拒绝这一条任务对应的 Promise
释放并补位.finally(...)无论成功失败，都减少计数并继续调度

run() 有两个入口：新任务入队时，以及已有任务结束时。这让队列既能在首次提交时启动，也能在槽位释放后继续推进。

为什么这里用 while

while (running < max && queue.length > 0) {
  // 取出一个任务，占用一个槽位，安排执行。
}

只要有空闲槽位和等待任务，就持续安排执行，直到槽位用满或队列为空。循环里没有 await，因此不会等当前任务完成才安排下一个。

它也不会一次启动所有任务：每安排一条，running 就同步加一，达到上限后循环停止。

在每次只添加一条、每次只释放一个槽位的简单流程中，if 也可能正常工作；while 更直接地表达了“把所有可用槽位补满”。

为什么把 task 放进 Promise 链

Promise.resolve().then(task) 会在 Promise 回调中调用 task。任务正常返回一个值、返回 Promise，或者同步抛错，都可以沿着后面的成功或失败分支处理。

如果改成 Promise.resolve(task())，会先执行 task() 再调用 Promise.resolve，同步异常就可能跳过后面的链式清理。这里把函数本身交给 .then()，统一捕获错误。

要传入“尚未执行的任务函数”，例如 () => fetch('/api/item')。如果传入的是已经发出的请求，调度器就无法控制它何时开始。

为什么需要 finally

只有成功时减少 running 会导致失败任务一直占着槽位，后续任务可能卡在队列里。放在 finally 中，无论结果如何都能释放槽位，并再次调用 run()。

失败只会拒绝当前任务对应的 Promise，不会停止整个队列。调用方仍需处理返回 Promise 的拒绝，例如添加 .catch()。

第二种：槽位已满才入队

createScheduler2 的调用方式相同，只需把示例中的工厂函数换成 createScheduler2(2)。

每次调用 addTo(task)，先创建一个 runTask。这个包装函数通过闭包记住当前任务的 task、resolve 和 reject，内部负责占位、执行、通知结果，以及结束后补位。

if (running < max) {
  runTask()
} else {
  queue.push(runTask)
}

有空位时直接调用 runTask()，没有空位时才排队。这里“直接调用”仍然只是同步增加 running、安排 Promise 链；真正的 task() 同样在微任务里执行。

例如上限为 2，依次提交 A、B、C：A、B 直接获得槽位，队列里只留下 C 的 runTask。B 结束后，先把 running 从 2 减到 1，再取出 C 的包装函数执行，计数立即回到 2。

为什么队列里放 runTask

排队的不只是原始任务，还需要记住任务结果应该交给谁。第一版用对象显式保存 { task, resolve, reject }，第二版让 runTask 的闭包保存这三者，因此出队后直接调用即可。

为什么这里用 if 就够了

每个任务结束只释放一个槽位，所以只需要取出一个等待任务补上。running-- 和 nextTask() 之间没有异步等待，补位函数又会同步执行 running++，不会让多个等待任务抢占同一个槽位。

这依赖当前实现的前提：并发上限固定、每次完成释放一个槽位。如果以后支持动态调高上限，再考虑集中调度、一次补满多个槽位。

两种写法怎么对照

对比点统一入队：createScheduler满额才入队：createScheduler2
队列内容保存任务及 resolve、reject 的对象闭包保存这些信息的 runTask 函数
提交任务先入队，再调用 run有空位直接安排，否则入队
完成后调用 run，用 while 补满可用槽位取出一个 runTask，补上刚释放的槽位
阅读重点调度逻辑集中在 run单个任务的执行和清理集中在 runTask

在这里的固定并发场景中，两版都能限制并发、按顺序启动等待任务，并在失败后继续推进队列。主要区别是代码组织方式；可以根据自己更容易复述哪种流程来选择。

需要记住的边界

  • max 约定为正整数，task 约定为函数；此实现没有额外参数校验。max = 0 时，任务会一直排队。
  • 队列保证等待任务按先入先出启动，不保证按提交顺序完成。
  • 如果某个任务永远不结束，它就会一直占用槽位；这里没有超时或取消功能。
  • 每次添加任务返回独立的 Promise；与可重启任务共用一个外层 Promise、只采纳最新轮次的设计不同。

面试时怎么讲

我用闭包保存等待队列和运行计数，每次添加任务都创建一个 Promise，把任务与它的 resolve、reject 一起入队。调度函数通过 while 在并发上限内取任务，再放进 Promise 链执行。成功或失败分别通知对应调用方，finally 统一释放槽位并继续调度，因此单个任务失败不会阻塞后面的任务。

第二种可以这样讲：我把单个任务的执行和清理封装成 runTask，用闭包保存任务及其 Promise 完成函数。有空位就直接安排执行，槽位满了才把 runTask 入队。任务结束时在 finally 释放一个槽位，再取出一个等待任务补上。