javascript - 如何在不锁定主体流的情况下知道提取何时结束？

Question

我正在向 API 发出请求，但他们的服务器只允许一定数量的事件连接，因此我想限制正在进行的提取数量。就我的目的而言，只有当 HTTP 响应正文到达客户端时，提取才会完成(而不是正在进行)。

我想创建一个像这样的抽象:

const fetchLimiter = new FetchLimiter(maxConnections);
fetchLimiter.fetch(url, options); // Returns the same thing as fetch()

这将使事情变得更简单，但似乎无法知道其他代码使用的流何时结束，因为流在​​读取时被锁定。可以使用 ReadableStream.tee() 将流分成两部分，使用一个并将另一个返回给调用者(也可能用它构造一个 Response )，但这会降低性能，对吗？

Answer 1

由于 fetch 使用 Promise，因此您可以利用它来制作一个简单的队列系统。

这是我之前用于对基于 Promise 的内容进行排队的方法。它通过创建 Promise 并将其解析器添加到数组来将项目排入队列。当然，在 Promise 解决之前，await 会阻止调用任何后续的 Promise。

当一次获取完成后，我们所要做的就是获取下一个解析器并调用它来开始下一次获取。 promise 得到解决，然后获取开始!

最好的部分是，由于我们实际上并不使用 fetch 结果，因此不必担心必须克隆或其他任何事情......我们只需将其完整传递，这样你就可以在稍后的then或其他事情中使用它。

*编辑:由于在 fetch Promise 解析后正文仍在流式传输，因此我添加了第三个选项，以便您可以传入正文类型，并让 FetchLimiter 为您检索和解析正文.

These all return a promise that is eventually resolved with the actual content.

这样你就可以让 FetchLimiter 为你解析正文。我这样做是为了让它返回一个 [response, data] 数组，这样您仍然可以检查响应代码、 header 等内容。

就此而言，如果您需要执行更复杂的操作，例如根据响应代码解析正文的不同方法，您甚至可以传入回调或在 await 中使用的内容。

示例

我添加了注释来指示 FetchLimiter 代码的开始和结束位置...其余部分只是演示代码。

它使用了一个带有 setTimeout 的假 fetch，它将在 0.5-1.5 秒之间解析。它将立即启动前三个请求，然后事件将满，它将等待一个解决。

发生这种情况时，您将看到 promise 已解决的注释，然后队列中的下一个 promise 将开始，然后您将看到 中的 then for 循环解析。我添加了then只是为了让您可以看到事件的顺序。

(function() {
  const fetch = (resource, init) => new Promise((resolve, reject) => {
    console.log('starting ' + resource);
    setTimeout(() => {
      console.log(' - resolving ' + resource);
      resolve(resource);
    }, 500 + 1000 * Math.random());
  });

  function FetchLimiter() {
    this.queue = [];
    this.active = 0;
    this.maxActive = 3;
    this.fetchFn = fetch;
  }
  FetchLimiter.prototype.fetch = async function(resource, init, respType) {
    // if at max active, enqueue the next request by adding a promise
    // ahead of it, and putting the resolver in the "queue" array.
    if (this.active >= this.maxActive) {
      await new Promise(resolve => {
        this.queue.push(resolve); // push, adds to end of array
      });
    }
    this.active++; // increment active once we're about to start the fetch
    const resp = await this.fetchFn(resource, init);
    let data;
    if (['arrayBuffer', 'blob', 'json', 'text', 'formData'].indexOf(respType) >= 0)
      data = await resp[respType]();
    this.active--; // decrement active once fetch is done
    this.checkQueue(); // time to start the next fetch from queue
    return [resp, data]; // return value from fetch
  };

  FetchLimiter.prototype.checkQueue = function() {
    if (this.active < this.maxActive && this.queue.length) {
      // shfit, pulls from start of array. This gives first in, first out.
      const next = this.queue.shift();
      next('resolved'); // resolve promise, value doesn't matter
    }
  }

  const limiter = new FetchLimiter();
  for (let i = 0; i < 9; i++) {
    limiter.fetch('/mypage/' + i)
      .then(x => console.log(' - .then ' + x));
  }
})();

注意事项:

• 我不能 100% 确定在 promise 解决时正文是否仍在流式传输...这似乎是您担心的问题。但是，如果这是一个问题，您可以使用 Body mixin 方法之一，例如 blob 或 text 或 json，直到 body 才解决内容已完全解析 ( see here )

• 我故意将其保持得非常短(比如 15 行实际代码)，作为一个非常简单的概念证明。您需要在生产代码中添加一些错误处理，以便在 fetch 由于连接错误或其他原因而拒绝时，您仍然会减少事件计数器并开始下一个 fetch.

• 当然它也使用了async/await语法，因为它更容易阅读。如果您需要支持较旧的浏览器，您需要使用 Promises 进行重写，或者使用 babel 或等效工具进行转译。