Estoy tratando de agregar la capacidad de procesamiento por lotes a un generador js asíncrono. La idea es tener una función que envuelva un generador no por lotes. Esta función llamaría al siguiente método del generador varias veces para iniciar varias operaciones asíncronas al mismo tiempo, luego devolvería el primer valor teniendo cuidado de rellenar el objeto por lotes a medida que devuelve sus elementos al cliente. Los siguientes ejemplos demuestran el caso de trabajo que no usa el contenedor así como el caso del contenedor que produce resultados correctos, pero no da como resultado el comportamiento deseado de ejecución simultánea de las promesas por lotes.
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()Esta función llamaría al siguiente método del generador varias veces para iniciar varias operaciones asíncronas al mismo tiempo.
Desafortunadamente (?) no es así como funcionan los generadores asíncronos. Si yield x en una async function* , lo que realmente está sucediendo es
await yield await x (consulte los pasos 5 y 8.b de la operación abstracta AsyncGeneratorYield ). Esto significa que todas las acciones dentro del cuerpo del generador serán estrictamente secuenciales y es posible que no se superpongan solo porque el generador asíncrono se itera "más rápido". Cada llamada .next() que realiza en un generador asíncrono está, de hecho, en cola para hacer posible la await dentro del cuerpo del generador.
Sin embargo, puede lograr el comportamiento deseado utilizando un iterador síncrono que ofrece promesas.
Finalmente, tenga en cuenta un gran inconveniente de sus implementaciones (tanto createBatchedAsyncGenerato como batchAsyncGenerator ): no manejan correctamente los errores y pueden causar rechazos de promesas no manejados que bloqueen su aplicación. Consulte Esperando más de una operación de espera simultánea y ¿Alguna diferencia entre await Promise.all() y múltiples await? para detalles.
La función de procesamiento por lotes podría ser una función de generador en sí misma:
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 // (eg 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)El generador de lotes se vuelve a crear en cada paso, hasta que se cierra la función del generador de envoltura.