使用嵌套的 Promise 创建一个 Promise 队列

Question

我正在实现一个大量获取和处理请求的查询引擎。我正在使用异步/等待。

现在执行流程在一个层次结构中运行，其中有一个包含查询的项目列表，每个查询都有一个提取。

我要做的是将项目捆绑在 n 组中，因此即使每个项目都有 m 个查询，其中包含提取，也只有 n*m 个请求同时运行；特别是同一域只会同时发出一个请求。

问题是，当我等待项目的执行时（在外部级别，有一段时间将项目分组并停止迭代，直到承诺解决），当内部查询的执行被推迟时，这些承诺正在解决因为 fetch 的内部等待。

这导致我的排队只是暂时停止，而不是等待内部承诺解决。

这是外部的排队类：

class AsyncItemQueue {
  constructor(items, concurrency) {
    this.items = items;
    this.concurrency = concurrency;
  }

  run = async () => {
    let itemPromises = [];

    const bundles = Math.ceil(this.items.length / this.concurrency);
    let currentBundle = 0;

    while (currentBundle < bundles) {
      console.log(`<--------- FETCHING ITEM BUNDLE ${currentBundle} OF ${bundles} --------->`);

      const lowerRange = currentBundle * this.concurrency;
      const upperRange = (currentBundle + 1) * this.concurrency;

      itemPromises.push(
        this.items.slice(lowerRange, upperRange).map(item => item.run())
      );

      await Promise.all(itemPromises);

      currentBundle++;
    }
  };
}


export default AsyncItemQueue;

这是队列正在运行的简单项目类。我省略了多余的代码。

class Item {

// ...

  run = async () => {
    console.log('Item RUN', this, this.name);

    return await Promise.all(this.queries.map(query => {
      const itemPromise = query.run(this.name);
      return itemPromise;

    }));
  }
}

这是项目中包含的查询。每个项目都有一个查询列表。再次，一些代码被删除，因为它并不有趣。

class Query {

// ...


  run = async (item) => {
    // Step 1: If requisites, await.
    if (this.requires) {
      await this.savedData[this.requires];
    }

    // Step 2: Resolve URL.
    this.resolveUrl(item);

    // Step 3: If provides, create promise in savedData.
    const fetchPromise = this.fetch();

    if (this.saveData) {
      this.saveData.forEach(sd => (this.savedData[sd] = fetchPromise));
    }


    // Step 4: Fetch.
    const document = await fetchPromise;

    // ...
  }
}

AsyncItemQueue 中的 while 正确停止，但仅在执行流程到达 Query 中的第 3 步之前。一旦它到达那个 fetch（标准 fetch 函数的包装器），外部 promise 就会解析，并且我最终会同时执行所有请求。

我怀疑问题出在 Query 类的某个地方，但我不知道如何避免外部承诺的解决。

我尝试让Query类run函数返回文档，以防万一，但无济于事。

任何想法或指导将不胜感激。我将尝试回答有关代码的任何问题或在需要时提供更多信息。

谢谢！

PS：这是一个带有工作示例的代码框：https://codesandbox.io/s/goofy-tesla-iwzem

正如您在控制台退出中看到的那样，while 循环在提取完成之前进行迭代，并且它们都在同时执行。

Answer 1

我已经解决了。

问题出在AsyncItemQueue 类中。具体来说：

itemPromises.push(
  this.items.slice(lowerRange, upperRange).map(item => item.run())
);

这是将一个承诺列表推入列表中，因此，稍后：

await Promise.all(itemPromises);

在该列表中未找到任何等待的承诺（因为它包含更多列表，其中包含承诺）。

解决方案是将代码更改为：

await Promise.all(this.items.slice(lowerRange, upperRange).map(item => item.run()));

现在它运行良好。项目以 n 个批次运行，新批次在前一个批次完成之前不会运行。

我不确定这会对除我以外的任何人有所帮助，但我会将其留在这里，以防有一天有人发现类似的问题。感谢您的帮助。