并行异步迭代器 - 有可能吗？答案

【问题标题】：Parallel asynchronous iteraor - is it possible?并行异步迭代器 - 有可能吗？
【发布时间】：2020-02-06 14:24:04
【问题描述】：

现在我有以下代码：

import axios from 'axios'

const urls = ['https://google.com', 'https://yahoo.com']

async function* requests() {
  for (const url of urls) {
    yield axios.get(url)
  }
}

;(async () => {
  for await (const n of requests()) {
    console.log(n.config.url) // prints https://google.com and then https://yahoo.com
  }
})()

按原样，请求不会阻塞节点的单个线程，而是按顺序发生。我想知道是否可以更改代码以强制并行。

【问题讨论】：

Promise.all 能解决你的问题吗？
不确定你的意思。我将如何使用 Promise.all 来并行化事情？我还需要使用可用的响应，因为我们有内存限制。
在你的情况下，如果你 await Promise.all(requests()) 它将返回一个数组，其中包含 axios.get() 的已解决承诺，因此它将并行运行所有请求，并且 Promise.all() 在每个 Promise 时解析迭代器里面已经解决了
这还不够好，我无法像之前所说的那样将这些响应中的每一个都保存在内存中。我需要使用它们，因为它们可用，就像流一样。
您可以对它们进行批处理，而不是每个请求，而是每 n 个请求迭代一次。或者使用 rxjs

标签： javascript node.js async-await async-iterator

【解决方案1】：

“更简单”的 no-deps 方法是对它们进行批处理并使用 Promise.all 生成每个批次

import axios from 'axios'

const urls = [
  'https://jsonplaceholder.typicode.com/todos/1', 
  'https://jsonplaceholder.typicode.com/posts/1',
  'https://jsonplaceholder.typicode.com/users/1',
  'https://jsonplaceholder.typicode.com/comments/1'
]

async function* requests(batchSize = 1) {
  let batchedRequests = [];
  for (const url of urls) {
    batchedRequests.push(axios.get(url));
    if (batchedRequests.length === batchSize) {
      yield Promise.all(batchedRequests);
      batchedRequests = [];
    }
  }
  if (batchedRequests.length) { //if there are requests left in batch
    yield Promise.all(batchedRequests);
  }
}

;(async () => {
  for await (const batch of requests(2)) {
    batch.forEach(n => console.log(n.config.url)) // prints https://google.com and then https://yahoo.com
  }
})()

您可以使用rxjs 来获得类似的结果，具有可观察对象在灵活性方面的优势，但它是另一个库，如果您不熟悉反应式流，可能会更复杂。这是我在该主题上找到的详细帖子：https://medium.com/@ravishivt/batch-processing-with-rxjs-6408b0761f39

【讨论】：