nodejs异步在createReadStream中等待

本文介绍了nodejs异步在createReadStream中等待的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

我正在逐行读取CSV文件并在MongoDB中插入/更新.预期的输出将是1. console.log(row);2. console.log(cursor);3.console.log("stream");

I am reading a CSV file line by line and inserting/updating in MongoDB. The expected output will be 1. console.log(row); 2. console.log(cursor); 3.console.log("stream");

但是得到类似的输出1. console.log(row);console.log(row);console.log(row);console.log(row);console.log(row);............ ............2. console.log(cursor);3.console.log("stream");请让我知道我在这里想念的东西.

But getting output like 1. console.log(row); console.log(row); console.log(row); console.log(row); console.log(row); ............ ............ 2. console.log(cursor); 3.console.log("stream"); Please let me know what i am missing here.

const csv = require('csv-parser'); const fs = require('fs'); var mongodb = require("mongodb"); var client = mongodb.MongoClient; var url = "mongodb://localhost:27017/"; var collection; client.connect(url,{ useUnifiedTopology: true }, function (err, client) { var db = client.db("UKCompanies"); collection = db.collection("company"); startRead(); }); var cursor={}; async function insertRec(row){ console.log(row); cursor = await collection.update({CompanyNumber:23}, row, {upsert: true}); if(cursor){ console.log(cursor); }else{ console.log('not exist') } console.log("stream"); } async function startRead() { fs.createReadStream('./data/inside/6.csv') .pipe(csv()) .on('data', async (row) => { await insertRec(row); }) .on('end', () => { console.log('CSV file successfully processed'); }); }

推荐答案

在您的 startRead()函数中， await insertRec()不会停止更多数据 insertRec()正在处理时，数据事件从流动开始.因此，如果您不希望在 insertRec()完成之前运行下一个 data 事件，则需要暂停然后恢复流.

In your startRead() function, the await insertRec() does not stop more data events from flowing while the insertRec() is processing. So, if you don't want the next data event to run until the insertRec() is done, you need to pause, then resume the stream.

async function startRead() { const stream = fs.createReadStream('./data/inside/6.csv') .pipe(csv()) .on('data', async (row) => { try { stream.pause(); await insertRec(row); } finally { stream.resume(); } }) .on('end', () => { console.log('CSV file successfully processed'); }); }

仅供参考，如果 insertRec()失败，您还需要一些错误处理.

FYI, you also need some error handling if insertRec() fails.