javascript - createReadStream end 在数据完成处理之前触发

Question

我正在尝试执行以下操作:

逐行流式传输 csv 文件。

修改每一行中包含的数据。

流式传输和处理所有行后，完成并继续执行下一个任务。

问题是 .on("end")在 .on("data") 之前发生火灾完成处理每一行。我怎样才能得到 .on("end")在 .on("data") 之后开火是否已完成所有线的处理？

下面是我所说的一个简单的例子:

import parse from 'csv-parse'; 

var parser = parse({});

fs.createReadStream(this.upload.location)
.pipe(parser)
.on("data", line => {
  var num = Math.floor((Math.random() * 100) + 1);
  num = num % 3;
  num = num * 1000;
  setTimeout( () => { 
    console.log('data process complete');
  }, num);
})
.on("end", () => {
   console.log('Done: parseFile');
   next(null);
});

提前致谢。

Answer 1

我认为问题在于 setTimeout 事件监听器中的 data (或任何其他异步任务)。 end 在 data 之后触发，但即使在流触发 end 之后，异步任务也会导致它记录消息。
如果您取出 setTimeout ，那么您会看到它在 end 之前将所有消息记录在 data 中。您仍然可以执行异步任务，但可能会有一批在流结束后运行。
此代码有助于解释发生了什么:

const fs = require('fs')

const testFileName = 'testfile.txt'

fs.writeFileSync(testFileName, '123456789')

let count = 0
const readStream = fs.createReadStream(testFileName, {
  encoding: 'utf8',
  highWaterMark: 1  // low highWaterMark so we can have more chunks to observe
})
readStream.on('data', (data) => {
  console.log('+++++++++++processing sync+++++++++++++')
  console.log(data)
  console.log('+++++++++++end processing sync+++++++++++++')
  setTimeout(() => {
    console.log('-----------processing async-------------')
    console.log(data)
    console.log('-----------end processing async-------------')
  }, ++count * 1000)
})
readStream.on('end', () => {
  console.log('stream ended but still have async tasks doing their thing')
  fs.unlinkSync(testFileName)
})