JS异步/等待任务队列

Question

在我的JS应用中，我正在使用async / await功能。我想执行多个API调用，并希望将它们依次触发。换句话说，我想替换这个简单的方法：

    const addTask =异步（网址，选项）=> {
        返回等待获取（URL，选项）
    }

与更复杂的东西..像：

    让tasksQueue = []
    const addTask =异步（网址，选项）=> {
        taskQueue.push（{url，options}）
        ... //在队列中执行提取
        返回等待...
    }

处理异步收益的最佳方法是什么？

Answer 1

您可以使用Queue数据结构作为基础，并在子类中添加特殊行为。aQueue有两个方法的众所周知的接口enqueue()（添加新项目到末尾）和dequeue()（删除第一个项目）。在您的情况下，dequeue()等待异步任务。

特殊行为：

1. 每次新任务（例如fetch('url')）入队时，this.dequeue()都会被调用。
2. 有什么 dequeue()作用：
   1. 如果队列为空？return false（突破递归）
   2. 如果队列很忙？return false（上一个任务未完成）
   3. 别的 ？从队列中删除第一个任务并运行它
   4. 任务“完成”（成功或有错误）？递归调用dequeue()（2.），直到队列为空。

class Queue {
  constructor() { this._items = []; }
  enqueue(item) { this._items.push(item); }
  dequeue()     { return this._items.shift(); }
  get size()    { return this._items.length; }
}

class AutoQueue extends Queue {
  constructor() {
    super();
    this._pendingPromise = false;
  }

  enqueue(action) {
    return new Promise((resolve, reject) => {
      super.enqueue({ action, resolve, reject });
      this.dequeue();
    });
  }

  async dequeue() {
    if (this._pendingPromise) return false;

    let item = super.dequeue();

    if (!item) return false;

    try {
      this._pendingPromise = true;

      let payload = await item.action(this);

      this._pendingPromise = false;
      item.resolve(payload);
    } catch (e) {
      this._pendingPromise = false;
      item.reject(e);
    } finally {
      this.dequeue();
    }

    return true;
  }
}

// Helper function for 'fake' tasks
// Returned Promise is wrapped! (tasks should not run right after initialization)
let _ = ({ ms, ...foo } = {}) => () => new Promise(resolve => setTimeout(resolve, ms, foo));
// ... create some fake tasks
let p1 = _({ ms: 50, url: '??', data: { w: 1 } });
let p2 = _({ ms: 20, url: '??', data: { x: 2 } });
let p3 = _({ ms: 70, url: '??', data: { y: 3 } });
let p4 = _({ ms: 30, url: '??', data: { z: 4 } });

const aQueue = new AutoQueue();
const start = performance.now();

aQueue.enqueue(p1).then(({ url, data }) => console.log('%s DONE %fms', url, performance.now() - start)); //          = 50
aQueue.enqueue(p2).then(({ url, data }) => console.log('%s DONE %fms', url, performance.now() - start)); // 50 + 20  = 70
aQueue.enqueue(p3).then(({ url, data }) => console.log('%s DONE %fms', url, performance.now() - start)); // 70 + 70  = 140
aQueue.enqueue(p4).then(({ url, data }) => console.log('%s DONE %fms', url, performance.now() - start)); // 140 + 30 = 170

互动演示：

完整代码演示：https : //codesandbox.io/s/async-queue-ghpqm?file= /src/ index.js您可以在控制台和/或开发工具“性能”选项卡中播放并观看结果。这个答案的其余部分基于它。

解释：

enqueue()返回一个新的 Promise，将被解决（或拒绝）在一些点以后。这个Promise可用于处理async任务 Fn的响应。

enqueue()实际上是push()一个Object进入队列，它持有任务 Fn 和返回的 Promise控制方法。

自从拆包回来Promiseinsta。开始运行，this.dequeue()每次我们将新任务加入队列时都会调用它。

将一些performance.measure()添加到我们的task，我们可以很好地可视化我们的队列：

(*.gif 动画)

• 第一排是我们的队列实例
• 新入队的人tasks有一个“?? 等待..”时期（第 3 行）（< 1ms如果队列为空，可能是）
• 在某些时候它会出队并“？运行..”一段时间（第二行）

日志输出（console.table()）：

说明：第一个task是队列初始化后的enqueue()d 2.58ms。因为我们的队列是空的，所以没有?? waiting（0.04ms？~ 40?m）。任务运行时间13.88ms? 出队

类Queue是只是一个包装原生ArrayFn's！

您当然可以在一个课程中实现这一点。我只是想表明，您可以从已知的数据结构中构建您想要的内容。不使用 有一些很好的理由Array：

• 甲Queue数据结构由一个定义的接口的两个公共方法。使用 aArray可能会诱使其他人在其上使用本机Array方法，例如.reverse(),.. 这会破坏定义。
• enqueue()并且dequeue()比push()和更具可读性shift()
• 如果你已经有一个外部实现的Queue类，你可以从它扩展（可重用的代码）
• 可以更换项目Array中class Queue由其他数据结构：A“双向链表”，这将降低代码复杂度为Array.shift()从O（n）的[线性]至O（1）[恒定。（？比原生数组 Fn 更好的时间复杂度！）（？最终演示）

代码限制

这个AutoQueue类不限于async函数。它处理任何事情，可以这样调用await item[MyTask](this)：

1. let task = queue => {..}? 同步功能
2. let task = async queue => {..}? 异步函数
3. let task = queue => new Promise(resolve => setTimeout(resolve, 100) ? new Promise()

注意：我们已经用 调用了我们的任务await，其中await 将任务的响应包装到一个Promise. Nr 2.（异步函数），总是Promise自己返回 a ，并且await调用只是将 a 包装Promise到 other 中Promise，效率稍低。Nr 3. 很好。返回的承诺不会被包装await

这是异步函数的执行方式：（来源）

1. 异步函数的结果总是一个 Promise p。该 Promise 是在开始执行异步函数时创建的。
2. 尸体被处决。执行可以通过返回或抛出永久完成。或者通过await暂时结束；在这种情况下，执行通常会在稍后继续。
3. Promisep被返回。

下面的代码演示了它是如何工作的：

async function asyncFunc() {
    console.log('asyncFunc()'); // (A)
    return 'abc';
}
asyncFunc().
    then(x => console.log(`Resolved: ${x}`)); // (B)
console.log('main'); // (C)

// Output:
// asyncFunc()
// main
// Resolved: abc

您可以依赖以下顺序：

1. 第(A)行：异步函数同步启动。异步函数的 Promise 是通过 return 解决的。
2. 行 (C)：继续执行。
3. 第 (B) 行：Promise 解析通知异步发生。

阅读更多：“可调用值”阅读更多：“异步函数”

性能限制

由于AutoQueue被限制在处理一个任务前一后，它可能会成为我们的应用程序的瓶颈。限制因素是：

1. 每次任务： ? 新enqueue()任务的频率。
2. 每个任务的运行时间? 阻塞时间dequeue()直到任务完成

1. 每次任务

这是我们的责任！我们可以随时获取当前的大小queue：size = queue.size。对于稳定增长的队列，您的外部脚本需要一个“故障转移”案例（检查“堆叠wait时间”部分）。

您想避免像这样的“队列溢出”，其中平均值/平均值waitTime随着时间的推移而增加。

+-------+----------------+----------------+----------------+----------------+
| tasks | enqueueMin(ms) | enqueueMax(ms) | runtimeMin(ms) | runtimeMax(ms) |
| 20    |              0 |            200 |             10 |             30 |
+-------+----------------+----------------+----------------+----------------+

• ? 任务20/20等待195ms直到 exec 开始
• ? 从我们的最后一个任务随机入队的时间开始，又需要+ ~232ms，直到所有任务都被解决。

2. 每个任务的运行时间

这个比较难处理。（等待 afetch()无法改善，我们需要等到 HTTP 请求完成）。
也许您的fetch()任务依赖于彼此的响应，并且长时间运行会阻止其他任务。

但是我们可以做一些事情：

• 也许我们可以缓存响应？减少下一个入队的运行时间。

• 也许我们fetch()来自 CDN 并且有一个我们可以使用的替代 URI。在这种情况下，我们可以new Promise从task将在下一个task是enqueue()d之前运行的我们返回一个。（参见“错误处理”）：

  queue.enqueue(queue => Promise.race(fetch('url1'), fetch('url2')));
  

• 也许您有某种无法缓存的“长轮询”或task每 x 秒运行一次的定期 ajax 。即使您不能减少运行时间本身，您也可以记录运行时间，这会给您一个近似值。下一次运行的估计。也许可以将长时间运行的任务交换到其他队列实例。

均衡 AutoQueue

什么是“高效” Queue？- 你的第一个想法可能是这样的：

最有效的在最短的时间内Queue处理最多tasks？

既然我们不能改善我们的task运行时间，我们可以减少等待时间吗？的例子是一个queue具有零（~0ms任务之间）的等待时间。

提示：为了比较我们的下一个例子，我们需要一些不会改变的基本统计数据：

+-------+----------------+----------------+------------------+------------------+
| count |  random fake runtime for tasks  |  random enqueue() offset for tasks  |
+-------+----------------+----------------+------------------+------------------+
| tasks | runtimeMin(ms) | runtimeMax(ms) | msEnqueueMin(ms) | msEnqueueMax(ms) |
| 200   |             10 |             30 |                0 |             4000 |
+-------+----------------+----------------+------------------+------------------+
     Avg. task runtime: ? (10ms + 30ms) / 2   = 20ms
     Total time:        ?  20ms * 200         = 4000ms ? 4s
  ? We expect our queue to be resolved after ~4s
  ? For consistent enqueue() frequency we set msEnqueueMax to 4000

• AutoQueue最后dequeue()完成~4.12s（^^见工具提示）。
• 这~120ms比我们预期的要长4s：

  提示：在每个任务之后都有一个小的“日志”块~0.3ms，在那里我构建/推送一个Object带有日志标记的日志到最后的全局“数组” console.table()。这解释了200 * 0.3ms = 60ms..丢失的60ms部分未被跟踪（你看到小任务之间的差距）-> 0.3ms/task 用于我们的测试循环，并且可能由于打开的 Dev-Tools 有一些延迟，..

我们稍后再回到这些时间。

我们的初始化代码queue：

const queue = new AutoQueue();
// .. get 200 random Int numbers for our task "fake" runtimes [10-30]ms
let runtimes = Array.from({ length: 200 }, () => rndInt(10, 30));
let i = 0;
let enqueue = queue => {
    if (i >= 200) {
        return queue; // break out condition
    }
    i++;
    queue
        .enqueue(
            newTask({ // generate a "fake" task with of a rand. runtime
                ms: runtimes[i - 1],
                url: _(i)
            })
        )
        .then(payload => {
            enqueue(queue);
        });
};
enqueue(queue); // start recurion

在上enqueue()一个任务完成之后，我们递归地执行下一个任务。您可能已经注意到了比喻为典型的 Promise.then()链条，对不对？

提示：Queue如果我们已经知道顺序tasks 运行的顺序和总数，我们就不需要 a 。我们可以使用Promise链并获得相同的结果。

有时我们在脚本开始时并不知道所有后续步骤。

..您可能需要更大的灵活性，我们要运行的下一个任务取决于前一个task. - 也许您的应用程序依赖于 REST API（多个端点），并且您被限制为最大 X 个同时 API 请求。我们不能向 API 发送来自整个应用程序的请求。你甚至不知道下一个请求什么时候得到enqueue()d（例如 API 请求是由click()事件？..

好的，对于下一个示例，我稍微更改了初始化代码：

我们现在在 [0-4000 毫秒] 期间内随机排列200 个任务。- 公平地说，我们将范围减少了30ms（最大任务运行时间）到 [0-3970 毫秒]。现在我们随机填充的队列有机会保持在4000ms limit 内。

我们可以得到什么或 Dev-Tools 性能登录：

• 随机enqueue()导致大量“等待”任务。

  有道理，因为我们将所有任务排在 first 中~4000ms，它们必须以某种方式重叠。检查表输出，我们可以验证：Maxqueue.size是22在任务170/200入队时。

• 等待任务分布不均。启动后甚至还有一些空闲部分。

  由于随机性，我们的第一个任务enqueue()不太可能获得0ms偏移量。 ~20ms每个任务的运行时间会随着时间的推移导致堆叠效应。

• 我们可以通过“等待毫秒”（见屏幕）对任务进行排序：最长等待时间是>400ms.

  queue.size（列：）sizeOnAdd和wait ms（见下一节）之间可能存在关系。

• 我们在初始化后AwaitQueue完成了最后一个dequeue() ~4.37s（检查“性能”选项卡中的工具提示）。20,786ms / task(expected: 20ms)的平均运行时间为我们提供了(expected: ? )的总运行时间。 4157.13ms4000ms4s

  我们仍然有我们的“日志”块和 exec。我们的测试脚本它自己的时间 ~120ms。还~37ms长吗？一开始就总结所有空闲的“差距”解释了缺失的~37ms

回到我们最初的“定义”

最有效的在最短的时间内Queue处理最多tasks？

假设：除了上例中的随机偏移，tasksget enqueue()d，两个队列在同一时间段内处理了相同数量的tasks（等于avg.runtime）。无论是排队的等待时间还是taskqueue.size总运行影响。两者的效率一样吗？

由于 aQueue本质上会缩小我们的编码可能性，因此Queue如果我们谈论高效代码（每次任务），最好不要使用 a 。

队列帮助我们将异步环境中的任务整理成同步模式。这正是我们想要的。? “连续运行未知的任务序列”。

如果你发现自己问这样的问题：“如果一个新task的队列被排入一个已经填满的队列，那么我们等待结果的时间会因其他人的运行时间而增加。这会降低效率！”。那么你做错了：

• 您要么将不依赖（以某种方式）彼此（逻辑或程序依赖）的任务排入队列，要么存在不会增加我们脚本的总运行时间的依赖关系。- 无论如何，我们必须等待其他人。

堆叠 wait次数

我们已经看到任务运行之前的高峰wait时间461.05ms。如果我们可以wait在决定将任务排队之前预测任务的时间，那不是很好吗？

首先，我们分析我们AutoQueue班级在更长时间内的行为。（重新发布屏幕）

我们可以从console.table()输出中构建图表：

在 a 的wait时间旁边task，我们可以看到随机 [10-30ms]runtime和 3 条曲线，代表当前queue.size，记录在 a task..

• .. 是 enqueued()
• .. 开始运行。( dequeue())
• .. 任务完成（就在下一个之前dequeue()）

再运行 2 次进行比较（类似趋势）：

• 图表运行 2：https : //i.imgur.com/EjpRyuv.png
• 图表运行 3：https : //i.imgur.com/0jp5ciV.png

我们可以找到彼此之间的依赖关系吗？

如果我们能找到这些记录的图表线中的任何一条之间的关系，它可能会帮助我们了解 aqueue随时间推移的行为（？不断充满新任务）。

Exkurs：什么是关系？我们正在寻找一个公式项目的wait ms曲线上的一个3个queue.size记录。这将证明两者之间存在直接依赖关系。

对于上次运行，我们更改了启动参数：

• 任务数： 200 ? 1000 (5x)

• msEnqueueMax: 4000ms? 20000ms(5x)

  +-------+----------------+----------------+------------------+------------------+
  | count |  random fake runtime for tasks  |  random enqueue() offset for tasks  |
  +-------+----------------+----------------+------------------+------------------+
  | tasks | runtimeMin(ms) | runtimeMax(ms) | msEnqueueMin(ms) | msEnqueueMax(ms) |
  | 1000  |             10 |             30 |                0 |            20000 |
  +-------+----------------+----------------+------------------+------------------+
       Avg. task runtime: ? (10ms + 30ms) / 2   = 20ms (like before)
       Total time:        ?  20ms * 1000        = 20000ms ? 20s
    ? We expect our queue to be resolved after ~20s 
    ? For consistent enqueue() frequency we set msEnqueueMax to 20000
  

（互动图表：https : //datawrapper.dwcdn.net/p4ZYx/2/）

我们看到了同样的趋势。wait ms随着时间的推移而增加（没什么新鲜的）。由于queue.size底部的3条线绘制在同一个图表中（Y 轴有ms刻度），因此它们几乎不可见。快速切换到对数刻度以进行更好的比较：

（互动图表：https : //datawrapper.dwcdn.net/lZngg/1/）

两条虚线的queue.size [on start]和queue.size [on end]几乎相互重叠，并下降至“0”一旦我们的队列变空，在最后。

queue.size [on add]看起来非常相似的wait ms线。这就是我们所需要的。

   {queue.size [on add]} * X = {wait ms}
? X = {wait ms} / {queue.size [on add]}

这本身在运行时对我们没有帮助，因为wait ms对于新的排队任务（尚未运行）来说是未知的。所以我们仍然有 2 个未知变量：X和wait ms。我们需要另一种关系来帮助我们。

首先，我们将我们的新配给打印{wait ms} / {queue.size [on add]}到图表中（浅绿色）及其平均值（浅绿色水平虚线）。这非常接近20ms(avg.run ms我们的任务），对吧？

切换回linearY 轴并将其“最大比例”设置为80ms以获得更好的视图。（提示：wait ms现在超出​​了视口）

（互动图表：https : //datawrapper.dwcdn.net/Tknnr/4/）

回到我们任务的随机运行时间（点云）。我们仍然有我们的“总平均值” 20.72ms（深绿色虚线水平）。我们还可以在运行时计算我们之前任务的平均值（例如，任务 370 入队？任务 [1,.., 269] 的当前平均运行时间是多少 = 平均运行时间）。但我们甚至可以更精确：

我们排队的任务越多，它们对总“平均运行时间”的影响就越小。因此，让我们计算最后一个例如 50的“平均运行时间” tasks。这导致了一致的影响每个任务对“平均运行时间”1/50。? 峰值运行时间变直并考虑趋势（上升/下降）。（深绿色水平路径曲线旁边来自我们 1. 方程的浅绿色）。

我们现在可以做的事情：

1. 我们可以从我们的第一个方程（浅绿色）中消除 X。? X可以用“之前的平均运行时间来表示，n例如 50 个任务（深绿色）。我们的新方程只取决于在运行时已知的变量，就在排队点：

   // mean runtime from prev. n tasks:
   X = {[taskRun[-50], .. , taskRun[-2], taskRun[-1] ] / n } ms
   
   // .. replace X in 1st equation:
   ? {wait ms} = {queue.size [on add]} * {[runtime[-50], .. , runtime[-2], runtime[-1] ] / n } ms
   

2. 我们

Answer 2

您可以保存上一个未完成的承诺，在调用next之前等待它fetch。

// fake fetch for demo purposes only
const fetch = (url, options) => new Promise(resolve => setTimeout(resolve, 1000, {url, options}))

// task executor
const addTask = (() => {
  let pending = Promise.resolve();
  
  const run = async (url, options) => {
    try {
      await pending;
    } finally {
      return fetch(url, options);
    }
  }

  // update pending promise so that next task could await for it
  return (url, options) => (pending = run(url, options))
})();

addTask('url1', {options: 1}).then(console.log)

addTask('url2', {options: 2}).then(console.log)

addTask('url3', {options: 3}).then(console.log)