尋找JS異步生成器的批處理方法

我正在嘗試向異步 js 生成器添加批處理功能。這個想法是有一個可以環繞非批處理生成器的函式。此函式將多次呼叫生成器的 next 方法以同時啟動多個異步操作，然后它將回傳第一個值，并在將其專案回傳給客戶端時注意重新填充批處理物件。以下示例演示了不使用包裝器的作業案例以及產生正確結果但不會導致批處理承諾的并發執行所需行為的包裝器案例。

function sleep(ms) {
  return new Promise(resolve => setTimeout(resolve, ms));
}

async function foo(v) {
  await sleep(1000);
  return v;
}

const createAsyncGenerator = async function*(){
  for (let i = 0; i < 5000; i  ) {
    yield foo(i);
  }
}

const createBatchedAsyncGenerator = async function*(batch_size){
  const batch = [];
  for (let i = 0; i < batch_size; i  ) {
    batch.push(foo(i));
  }
  for (let i = batch_size; i < 500; i  ) {
    batch.push(foo(i));
    yield batch.shift();
  }
}


function batchAsyncGenerator(generator) {
  return {
    batch: [],
    [Symbol.asyncIterator]() {
      while (this.batch.length < 5) {
        this.batch.push(generator.next());
      }
      return {
        batch: this.batch,
        async next() {
          this.batch.push(generator.next());
          const result = this.batch.shift();
          return result;
        }
      }
    }
  }
}

const batching_works = async () => {
  const asyncGenerator = createBatchedAsyncGenerator(5);
  for await (const item of asyncGenerator) {
    console.log(item)
  }
}

const batching_doesnt_work = async () => {
  const asyncGenerator = batchAsyncGenerator(createAsyncGenerator());
  for await (const item of asyncGenerator) {
    console.log(item)
  }
}

batching_works()
//batching_doesnt_work()

uj5u.com熱心網友回復：

該函式會多次呼叫生成器的 next 方法以同時啟動多個異步操作

不幸的是（？）不是異步生成器的作業方式。如果你yield x在一個async function*，實際發生的是

await yield await x

（參見AsyncGeneratorYield抽象操作的步驟 5 和 8.b ）。這意味著生成器主體內的所有操作都將是嚴格順序的，并且可能不會僅僅因為異步生成器“更快”地迭代而重疊。.next()實際上，您在異步生成器上進行的每個呼叫都會排隊以使await生成器主體內的 s 成為可能。

不過，您可以使用產生承諾的同步迭代器來實作所需的行為。

最后，請注意您的實作（createBatchedAsyncGenerato和batchAsyncGenerator）的一個巨大缺點：它們不能正確處理錯誤，并且可能會導致未處理的 Promise 拒絕使您的應用程式崩潰。請參閱等待多個并發等待操作和await Promise.all() 和多個等待之間有什么區別？詳情。

uj5u.com熱心網友回復：

批處理函式本身可以是生成器函式：

function sleep(ms) {
  return new Promise(resolve => setTimeout(() => resolve(), ms));
}

const foo = async(v) => {
  await sleep(150)
  return v
}

const createAsyncGenerator = async function*() {
  for (let i = 0; i < 50; i  ) {
    const r = await foo(i)
    yield r
  }
}

// creating the batches & handling the results
const asyncBatchGenerator = async function*({
  batch,
  fn
}) {
  const a = [...Array(batch)].fill('').map((_) => fn.next())
  const res = await Promise.all(a)
  yield res
}

// wrapper function that handles the different states
// (e.g. asyncBatchGenerator is re-initialized, while fn
// is kept until "closed"
const batchingWrapper = async(fn) => {
  const genFn = fn()
  const bf = () => asyncBatchGenerator({
    batch: 8,
    fn: genFn
  })
  let res = []
  let genClosed = false

  while (!genClosed) {
    for await (let value of bf()) {
      // creating the return array: only items that
      // hold a real value are added to the result
      res = [...res, ...value.filter(({
        done
      }) => !done)]
      // logging the growing return array:
      console.log(res)

      if (value.some(({
          done
        }) => done)) {
        genClosed = true
      }
    }
  }
  console.log('generator closed')
}

// calling the wrapper with the argument:
batchingWrapper(createAsyncGenerator)

批處理生成器在每次傳遞時都會重新創建 - 直到包裝生成器函式關閉。

標籤：

上一篇：防止同時呼叫同一個Promise/async函式

下一篇：更改按鈕引數的最佳方法是什么C#