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

Question

现在我有以下代码：

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
  }
})()

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

Answer 1

“更简单”的无依赖方式是对它们进行批处理，并用

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

Answer 2

您可以使用

lfi

，它支持“并发迭代”：

pipe(
  asConcur(urls),
  mapConcur(url => axios.get(url)),
  forEachConcur(response => console.log(n.config.url)),
)

每个 URL 将在操作管道中移动，而不会被其他 URL 阻塞。