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开始!

最好的部分，因为我们实际上并不消耗 fetch结果，不必担心 clone或任何东西...我们只是完整地传递它，以便您可以在以后使用它 then或者其他的东西。

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

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

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

就此而言，你甚至可以传入一个回调或其他东西在 await 中使用。如果您需要做一些更复杂的事情，例如根据响应代码解析正文的不同方法。

示例

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

它使用的是假的 fetch使用 setTimeout，它将在 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 是否仍在流动......这似乎是你的问题。但是，如果这是一个问题，您可以使用其中一种 Body mixin 方法，例如 blob或 text或 json ，直到正文内容完全解析( see here )

才能解析

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

当然它也在使用 async/await语法，因为它更容易阅读。如果您需要支持较旧的浏览器，则需要使用 Promises 重写或使用 babel 或等效程序进行转换。