我认为我没有正确有效地实现和使用可读流？

Question

它是一个从文本文件'IN.txt'中读取数据并以json格式写入'copy.json'文件的程序。
在文本文件的每一行中，单词由制表符分隔，并且使用制表符将行拆分为数组。

我认为以这种方式实现可读流会一次又一次地覆盖相同的数据，这对于大文件来说效率不高。
我确实尝试了很多不同的方法，但我遇到了内存泄漏、_read 方法未定义等错误。

const fs = require('fs');
const readLine = require('readline');
const { Readable } = require('stream');
const dataArray = [];

//creating readline interface
const lineReader = readLine.createInterface({
    input: fs.createReadStream(__dirname + '/IN.txt'),
});

const fields = ['country', 'pin', 'place', 'state', 'code', 'division', 'admin', 'mandal', 'xxx', 'lat', 'long'];

//reading data from text file line by line and spliting each line into array
lineReader.on('line', function (line) {
    let words = line.split('\t');
    writeToFile(fields, words);
});

lineReader.on('close', function (line) {
    console.log('***Finished***');
    process.exit(0);
});

//words array will be like ["IN","744301", "Mus Andaman & Nicobar Islands", "01 Nicobar 638 Carnicobar" , "9.2333", "92.7833","4"]
//creating obj with fields and words array and pushing into array
function writeToFile(fields, words) {
    var obj = {};
    for(let i = 0; i < fields.length; i++) {
        obj[fields[i]] = words[i];
    }
    dataArray.push(obj);
    //implementing readable stream and pushing string into it 
    const rStream = new Readable();
    rStream.push(JSON.stringify(dataArray, null, 4));
    rStream.push(null);
    const output = fs.createWriteStream(__dirname + '/copy.json');
    //piping to output
    rStream.pipe(output);
}

这是 IN.txt 文件的小快照

IN.txt file

Answer 1

在每次调用 writeToFile（基本上是读取每一行）时，您都在创建一个 readStream 并将 dataArray 复制到它，并通过管道传输到一个写入流。如果您已经对文件打开了读取流，则不需要此操作。

很好的文字阅读：https://medium.freecodecamp.org/node-js-streams-everything-you-need-to-know-c9141306be93

试一试： process.memoryUsage().heapUsed / 1024 / 1024 给了我大约 147 MB​​ 的内存堆，用于大约 14 MB 的 IN.txt 文件。

const fs = require('fs');
const readLine = require('readline');
const { Readable } = require('stream');
const output = fs.createWriteStream(__dirname + '/copy.json');
const dataArray = [];

//creating readline interface
const lineReader = readLine.createInterface({
    input: fs.createReadStream(__dirname + '/IN.txt')
});

const fields = ['country', 'pin', 'place', 'state', 'code', 'division',     'admin', 'mandal', 'xxx', 'lat', 'long'];

//reading data from text file line by line and pushing it to an array
lineReader.on('line', function (line) {
    let words = line.split('\t');
    dataArray.push(getLineContent(fields, words));
});

lineReader.on('close', function (line) {
    console.log('***Finished***');
    output.write(JSON.stringify(dataArray, null, 4));
    output.end();
    process.exit(0);
});

//words array will be like ["IN","744301", "Mus Andaman & Nicobar Islands", "01 Nicobar 638 Carnicobar" , "9.2333", "92.7833","4"]
//creating obj with fields and words
function getLineContent(fields, words) {
    var obj = {};
    for(let i = 0; i < fields.length; i++) {
        obj[fields[i]] = words[i];
    }
    return obj;
}

更高效的解决方案：

process.memoryUsage().heapUsed / 1024 / 1024 为大约 14 MB 的 IN.txt 文件提供了大约 5-7 MB 的内存堆（与上述方法相比有了显着改进）。

更多参考文字：

1. Stream highWaterMark misunderstanding
2. Pausing readline in Node.js
3. https://www.valentinog.com/blog/memory-usage-node-js/

以下内容可能有助于您快速入门：

const fs = require('fs');
const readLine = require('readline');
const { Readable } = require('stream');
const output = fs.createWriteStream(__dirname + '/copy.json');

//creating readline interface
const lineReader = readLine.createInterface({
    input: fs.createReadStream(__dirname + '/IN.txt')
});

const fields = ['country', 'pin', 'place', 'state', 'code', 'division',         'admin', 'mandal', 'xxx', 'lat', 'long'];

let lineCount = 0;
let writeAllowed = true; //Turns to false when stream.write starts     returning false
let paused = false; //Pause Readline
let buffstr = ""; //To handle leaks after calling readLine pause()

//reading data from text file line by line and pushing it to an array
lineReader.on('line', function (line) {
    lineCount++;
    let words = line.split('\t');
    let lineJson = getLineContent(fields, words);

    if (paused) {
      if(lineCount > 1) {
        buffstr = buffstr + ",";
      }
      buffstr = buffstr + JSON.stringify(lineJson, null, 4);
    }
    else {
      if(!writeAllowed) {
        lineReader.pause();
      }
      lineCount === 1 ? writeMe('[') : writeMe(",");
      writeMe(JSON.stringify(lineJson, null, 4));
    }
});

lineReader.on('pause', function() {
   paused = true;
});

lineReader.on('resume', function() {
   paused = false;
});

lineReader.on('close', function (line) {
    output.write(buffstr);
    output.write(']');
    output.end();
    console.log(`***Finished*** Memory heap used:     ${process.memoryUsage().heapUsed / 1024 / 1024} MB`);
});

function writeMe(str){
   if(writeAllowed){
      writeAllowed = writeAllowed && output.write(str);
   }
   else{
      buffstr += str;
      output.once('drain', function() {
         lineReader.resume();
         output.write(buffstr);
         buffstr = ""; //Possible scope of improvement. Need to check if     any race condition
         writeAllowed = true;
      });
   }
}

//words array will be like ["IN","744301", "Mus Andaman & Nicobar     Islands", "01 Nicobar 638 Carnicobar" , "9.2333", "92.7833","4"]
//creating obj with fields and words
function getLineContent(fields, words) {
    var obj = {};
    for(let i = 0; i < fields.length; i++) {
        obj[fields[i]] = words[i];
    }
    return obj;
}