是否可以让事件处理程序等到异步/基于Promise的代码完成？

编程问答 2022-08-12

问题描述

我正在nodejs模式下使用出色的Papa Parse库，将超过一百万行的大型（500 MB）CSV文件流式传输到慢速持久性API中，一次只能处理一个请求。持久性API基于Promise，但是从Papa Parse，我在 synchronous 事件中收到了每个解析的CSV行，如下所示：parseStream.on("data",row => { ... }

我面临的挑战是Papa Parse如此快地从流中转储其CSV行，以至于我的慢速持久性API无法跟上。因为Papa是同步且我的API是基于 Promise 的，所以我不能只在await doDirtyWork(row)事件处理程序中调用on，因为同步和异步代码不会混在一起。

或者他们可以混在一起，我只是不知道怎么做？

我的问题是，我可以让Papa的事件处理程序等待我的API调用完成吗？是直接在on("data")事件中执行持久性API请求，还是使on()函数以某种方式徘徊，直到脏API工作完成？

就内存占用量而言，到目前为止，我拥有的解决方案并不比使用Papa的非流模式好得多。实际上，我需要以生成器函数迭代的形式排队激流on("data")。我也可以在数组中排队诺言工厂，然后循环处理。无论哪种方式，我最终都会将几乎整个CSV文件保存为内存中大量将来的Promises（承诺工厂），直到我的慢速API调用一直有效为止。

async importCSV(filePath) {
    let parsednum = 0,processednum = 0;

    async function* gen() {
        let pf = yield;
        do {
            pf = yield await pf();
        } while (typeof pf === "function");
    };

    var g = gen();
    g.next();


    await new Promise((resolve,reject) => {
        try {
            const dataStream = fs.createReadStream(filePath);
            const parseStream = Papa.parse(Papa.NODE_STREAM_INPUT,{delimiter: ",",header: false});
            dataStream.pipe(parseStream);

            parseStream.on("data",row => {

                // Received a CSV row from Papa.parse()

                try {
                    console.log("PA#",parsednum,": parsed",row.filter((e,i) => i <= 2 ? e : undefined)
                    );
                    parsednum++;

                    // Simulate some really slow async/await dirty work here,for example 
                    // send requests to a one-at-a-time persistence API

                    g.next(() => {  // don't execute Now,call in sequence via the generator above
                        return new Promise((res,rej) => {
                            console.log(
                                "DW#",processednum,": dirty work START",i) => i <= 2 ? e : undefined)
                            );
                            setTimeout(() => {
                                console.log(
                                    "DW#",": dirty work STOP ",i) => i <= 2 ? e : undefined)
                                );
                                processednum++;
                                res();
                            },1000)
                        })
                    
                    });
                } catch (err) {
                    console.log(err.stack);
                    reject(err);                    
                }
            });
            parseStream.on("finish",() => {
                console.log(`Parsed ${parsednum} rows`);
                resolve();
            });

        } catch (err) {
            console.log(err.stack);
            reject(err);                    
        }
    });
    while(!(await g.next()).done);
}

那为什么要赶爸爸呢？为什么不让我慢一点地处理文件-原始CSV文件中的数据不会消失，我们有几个小时才能完成流传输，为什么还要用on("data")个事件来锤击我，似乎放慢了速度？

因此，我确实需要的是让Papa成为一名爷爷，并尽可能减少或消除CSV行的任何排队或缓冲。理想情况下，我将能够与我的API的速度（或缺乏速度）完全同步Papa的解析事件。因此，如果不是因为教条使异步代码无法使同步代码“休眠”，我理想地将每个CSV行发送到Papa事件中的 API ，然后仅 将控制权返回给爸爸。

建议？事件处理程序的某种“松散耦合”和我的异步API的缓慢也很好。我不介意几百行排队。但是当成千上万的东西堆积起来时，我很快就会耗尽。

解决方法

为什么要用on("data")的事件锤击我，但似乎无法放慢脚步？

可以，您只是没有要求爸爸停止。您可以先调用stream.pause()，然后再调用stream.resume()，以使用Node流的内置背压。

但是，有比使用自己的基于回调的代码use the stream as an async iterator更好的API了！当您在await循环的主体中for await时，生成器也必须暂停。所以你可以写

async importCSV(filePath) {
    let parsedNum = 0;

    const dataStream = fs.createReadStream(filePath);
    const parseStream = Papa.parse(Papa.NODE_STREAM_INPUT,{delimiter: ",",header: false});
    dataStream.pipe(parseStream);

    for await (const row of parseStream) {
        // Received a CSV row from Papa.parse()
        const data = row.filter((e,i) => i <= 2 ? e : undefined);
        console.log("PA#",parsedNum,": parsed",data);
        parsedNum++;
        await dirtyWork(data);
    }
    console.log(`Parsed ${parsedNum} rows`);
}

importCSV('sample.csv').catch(console.error);

let processedNum = 0;
function dirtyWork(data) {
    // Simulate some really slow async/await dirty work here,// for example send requests to a one-at-a-time persistence API
    return new Promise((res,rej) => {
        console.log("DW#",processedNum,": dirty work START",data)
        setTimeout(() => {
             console.log("DW#",": dirty work STOP ",data);
             processedNum++;
             res();
        },1000);
    });
}
,

JavaScript中的异步代码有时很难理解。记住Node操作如何处理并发很重要。

节点进程是单线程的，但是它使用了一个称为事件循环的概念。其结果是异步代码和回调实质上是同一事物的等效表示。

当然，您需要一个异步函数才能使用await，但是从Papa Parse进行的回调可以是一个异步函数：

parse.on("data",async row => {
  await sync(row)
})

await操作完成后，箭头函数结束，并且所有对row的引用都将被删除，因此垃圾收集器可以成功收集row，释放该内存。

这种效果是每次解析一行时并发执行sync，因此，如果一次只能同步一条记录，那么我建议将sync函数包装在一个去抖动器中。

backpressurees6-promisenode-streamspromise