nodejs异步在createReadStream中等待 [英] nodejs async await inside 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.

这篇关于nodejs异步在createReadStream中等待的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！