在node.js中一次读取一行文件？

Question

我正在尝试一次读取一个大文件。我找到了处理这个主题的a question on Quora，但我缺少一些联系以使整个事情融合在一起。

 var Lazy=require("lazy");
 new Lazy(process.stdin)
     .lines
     .forEach(
          function(line) { 
              console.log(line.toString()); 
          }
 );
 process.stdin.resume();

我想弄清楚的是如何从文件中一次读取一行，而不是像本示例中那样从 STDIN 读取。

我试过了：

 fs.open('./VeryBigFile.csv', 'r', '0666', Process);

 function Process(err, fd) {
    if (err) throw err;
    // DO lazy read 
 }

但它不起作用。我知道在紧要关头我可以回退到使用 PHP 之类的东西，但我想弄清楚这一点。

我不认为其他答案会起作用，因为文件比我运行它的服务器大得多。

Answer 1

从 Node.js v0.12 和 Node.js v4.0.0 开始，有一个稳定的 readline 核心模块。这是从文件中读取行的最简单方法，无需任何外部模块：

const fs = require('fs');
const readline = require('readline');

async function processLineByLine() {
  const fileStream = fs.createReadStream('input.txt');

  const rl = readline.createInterface({
    input: fileStream,
    crlfDelay: Infinity
  });
  // Note: we use the crlfDelay option to recognize all instances of CR LF
  // ('\r\n') in input.txt as a single line break.

  for await (const line of rl) {
    // Each line in input.txt will be successively available here as `line`.
    console.log(`Line from file: ${line}`);
  }
}

processLineByLine();

或者：

var lineReader = require('readline').createInterface({
  input: require('fs').createReadStream('file.in')
});

lineReader.on('line', function (line) {
  console.log('Line from file:', line);
});

即使没有最终的\n，也可以正确读取最后一行（从 Node v0.12 或更高版本开始）。

更新：这个例子是added to Node's API official documentation。

Answer 2

对于这样一个简单的操作，不应该对第三方模块有任何依赖。放轻松。

var fs = require('fs'),
    readline = require('readline');

var rd = readline.createInterface({
    input: fs.createReadStream('/path/to/file'),
    output: process.stdout,
    console: false
});

rd.on('line', function(line) {
    console.log(line);
});

Answer 3

您不必open 文件，而是必须创建一个ReadStream。

fs.createReadStream

然后将该流传递给Lazy

Answer 4

require('fs').readFileSync('file.txt', 'utf-8').split(/\r?\n/).forEach(function(line){
  console.log(line);
})

Answer 5

2019 年更新

Nodejs 官方文档中已经发布了一个很棒的示例。 here

这需要在您的机器上安装最新的 Nodejs。 >11.4

const fs = require('fs');
const readline = require('readline');

async function processLineByLine() {
  const fileStream = fs.createReadStream('input.txt');

  const rl = readline.createInterface({
    input: fileStream,
    crlfDelay: Infinity
  });
  // Note: we use the crlfDelay option to recognize all instances of CR LF
  // ('\r\n') in input.txt as a single line break.

  for await (const line of rl) {
    // Each line in input.txt will be successively available here as `line`.
    console.log(`Line from file: ${line}`);
  }
}

processLineByLine();

Answer 6

有一个非常好的模块用于逐行读取文件，它被称为line-reader

你只需写：

var lineReader = require('line-reader');

lineReader.eachLine('file.txt', function(line, last) {
  console.log(line);
  // do whatever you want with line...
  if(last){
    // or check if it's the last one
  }
});

如果您需要更多控制权，您甚至可以使用“java 样式”界面来迭代文件：

lineReader.open('file.txt', function(reader) {
  if (reader.hasNextLine()) {
    reader.nextLine(function(line) {
      console.log(line);
    });
  }
});

Answer 7

老话题，但这个可行：

var rl = readline.createInterface({
      input : fs.createReadStream('/path/file.txt'),
      output: process.stdout,
      terminal: false
})
rl.on('line',function(line){
     console.log(line) //or parse line
})

简单。无需外部模块。

Answer 8

您可以随时使用自己的线路阅读器。我还没有对这个 sn-p 进行基准测试，但它正确地将传入的块流拆分为没有尾随 '\n'

的行

var last = "";

process.stdin.on('data', function(chunk) {
    var lines, i;

    lines = (last+chunk).split("\n");
    for(i = 0; i < lines.length - 1; i++) {
        console.log("line: " + lines[i]);
    }
    last = lines[i];
});

process.stdin.on('end', function() {
    console.log("line: " + last);
});

process.stdin.resume();

我在处理需要在日志解析期间累积数据的快速日志解析脚本时确实想到了这个，我觉得尝试使用 js 和 node 而不是使用 perl 或 bash 来做这件事会很好。

无论如何，我确实觉得小型 nodejs 脚本应该是自包含的，而不是依赖第三方模块，所以在阅读了这个问题的所有答案之后，每个都使用各种模块来处理行解析，一个 13 SLOC 原生 nodejs 解决方案可能是感兴趣的。

Answer 9

用carrier module:

var carrier = require('carrier');

process.stdin.resume();
carrier.carry(process.stdin, function(line) {
    console.log('got one line: ' + line);
});

Answer 10

由于节点中的排水/暂停/恢复工作方式，我在尝试处理这些行并将它们写入另一个流时使用 Lazy 逐行读取，最终导致大量内存泄漏（请参阅：@987654321 @（我爱这个人顺便说一句））。我还没有仔细研究 Lazy 以了解确切原因，但我无法暂停我的读取流以允许在 Lazy 不退出的情况下进行排水。

我编写了将大量 csv 文件处理为 xml 文档的代码，您可以在此处查看代码：https://github.com/j03m/node-csv2xml

如果您使用 Lazy line 运行以前的修订版，它会泄漏。最新版本根本没有泄漏，您可能可以将其用作阅读器/处理器的基础。虽然我有一些定制的东西在那里。

编辑：我想我还应该注意，我的 Lazy 代码运行良好，直到我发现自己编写了足够大的 xml 片段，因为必要而耗尽/暂停/恢复。对于较小的块，这很好。

Answer 11

编辑：

使用transform stream。

---

使用BufferedReader，您可以读取行数。

new BufferedReader ("lorem ipsum", { encoding: "utf8" })
    .on ("error", function (error){
        console.log ("error: " + error);
    })
    .on ("line", function (line){
        console.log ("line: " + line);
    })
    .on ("end", function (){
        console.log ("EOF");
    })
    .read ();

Answer 12

自从发布我的原始答案后，我发现split 是一个非常易于使用的节点模块，用于在文件中读取行；它也接受可选参数。

var split = require('split');
fs.createReadStream(file)
    .pipe(split())
    .on('data', function (line) {
      //each chunk now is a seperate line! 
    });

尚未对非常大的文件进行测试。如果您这样做，请告诉我们。

Answer 13

在大多数情况下，这应该足够了：

const fs = require("fs")

fs.readFile('./file', 'utf-8', (err, file) => {
  const lines = file.split('\n')

  for (let line of lines)
    console.log(line)
});

Answer 14

我对缺乏一个全面的解决方案感到沮丧，所以我把我自己的尝试放在一起 (git / npm)。复制粘贴的功能列表：

• 交互式行处理（基于回调，不将整个文件加载到 RAM 中）
• （可选）返回数组中的所有行（详细或原始模式）
• 以交互方式中断流式传输，或执行类似映射/过滤的处理
• 检测任何换行约定 (PC/Mac/Linux)
• 正确的 eof/最后一行处理
• 正确处理多字节 UTF-8 字符
• 检索每行的字节偏移和字节长度信息
• 随机访问，使用基于行或基于字节的偏移量
• 自动映射线偏移信息，加快随机访问速度
• 零依赖
• 测试

美国国立卫生研究院？你决定:-)

Answer 15

function createLineReader(fileName){
    var EM = require("events").EventEmitter
    var ev = new EM()
    var stream = require("fs").createReadStream(fileName)
    var remainder = null;
    stream.on("data",function(data){
        if(remainder != null){//append newly received data chunk
            var tmp = new Buffer(remainder.length+data.length)
            remainder.copy(tmp)
            data.copy(tmp,remainder.length)
            data = tmp;
        }
        var start = 0;
        for(var i=0; i<data.length; i++){
            if(data[i] == 10){ //\n new line
                var line = data.slice(start,i)
                ev.emit("line", line)
                start = i+1;
            }
        }
        if(start<data.length){
            remainder = data.slice(start);
        }else{
            remainder = null;
        }
    })

    stream.on("end",function(){
        if(null!=remainder) ev.emit("line",remainder)
    })

    return ev
}


//---------main---------------
fileName = process.argv[2]

lineReader = createLineReader(fileName)
lineReader.on("line",function(line){
    console.log(line.toString())
    //console.log("++++++++++++++++++++")
})

Answer 16

我想解决同样的问题，基本上是 Perl 中的：

while (<>) {
    process_line($_);
}

我的用例只是一个独立的脚本，而不是服务器，所以同步很好。这些是我的标准：

• 可以在许多项目中重用的最小同步代码。
• 对文件大小或行数没有限制。
• 对行的长度没有限制。
• 能够处理 UTF-8 格式的完整 Unicode，包括 BMP 之外的字符。
• 能够处理 *nix 和 Windows 行尾（我不需要老式 Mac）。
• 要包含在行中的行结束字符。
• 能够处理带或不带行尾字符的最后一行。
• 不使用任何未包含在 node.js 发行版中的外部库。

这是一个让我了解 node.js 中的低级脚本类型代码并确定它作为 Perl 等其他脚本语言替代品的可行性的项目。

经过惊人的努​​力和几次错误的开始，这就是我想出的代码。它非常快，但比我预期的要简单：(fork it on GitHub)

var fs            = require('fs'),
    StringDecoder = require('string_decoder').StringDecoder,
    util          = require('util');

function lineByLine(fd) {
  var blob = '';
  var blobStart = 0;
  var blobEnd = 0;

  var decoder = new StringDecoder('utf8');

  var CHUNK_SIZE = 16384;
  var chunk = new Buffer(CHUNK_SIZE);

  var eolPos = -1;
  var lastChunk = false;

  var moreLines = true;
  var readMore = true;

  // each line
  while (moreLines) {

    readMore = true;
    // append more chunks from the file onto the end of our blob of text until we have an EOL or EOF
    while (readMore) {

      // do we have a whole line? (with LF)
      eolPos = blob.indexOf('\n', blobStart);

      if (eolPos !== -1) {
        blobEnd = eolPos;
        readMore = false;

      // do we have the last line? (no LF)
      } else if (lastChunk) {
        blobEnd = blob.length;
        readMore = false;

      // otherwise read more
      } else {
        var bytesRead = fs.readSync(fd, chunk, 0, CHUNK_SIZE, null);

        lastChunk = bytesRead !== CHUNK_SIZE;

        blob += decoder.write(chunk.slice(0, bytesRead));
      }
    }

    if (blobStart < blob.length) {
      processLine(blob.substring(blobStart, blobEnd + 1));

      blobStart = blobEnd + 1;

      if (blobStart >= CHUNK_SIZE) {
        // blobStart is in characters, CHUNK_SIZE is in octets
        var freeable = blobStart / CHUNK_SIZE;

        // keep blob from growing indefinitely, not as deterministic as I'd like
        blob = blob.substring(CHUNK_SIZE);
        blobStart -= CHUNK_SIZE;
        blobEnd -= CHUNK_SIZE;
      }
    } else {
      moreLines = false;
    }
  }
}

它可能会被进一步清理，这是反复试验的结果。

Answer 17

基于生成器的线路阅读器：https://github.com/neurosnap/gen-readlines

var fs = require('fs');
var readlines = require('gen-readlines');

fs.open('./file.txt', 'r', function(err, fd) {
  if (err) throw err;
  fs.fstat(fd, function(err, stats) {
    if (err) throw err;

    for (var line of readlines(fd, stats.size)) {
      console.log(line.toString());
    }

  });
});

Answer 18

如果您想逐行读取文件并将其写入另一个文件：

var fs = require('fs');
var readline = require('readline');
var Stream = require('stream');

function readFileLineByLine(inputFile, outputFile) {

   var instream = fs.createReadStream(inputFile);
   var outstream = new Stream();
   outstream.readable = true;
   outstream.writable = true;

   var rl = readline.createInterface({
      input: instream,
      output: outstream,
      terminal: false
   });

   rl.on('line', function (line) {
        fs.appendFileSync(outputFile, line + '\n');
   });
};

Answer 19

var fs = require('fs');

function readfile(name,online,onend,encoding) {
    var bufsize = 1024;
    var buffer = new Buffer(bufsize);
    var bufread = 0;
    var fd = fs.openSync(name,'r');
    var position = 0;
    var eof = false;
    var data = "";
    var lines = 0;

    encoding = encoding || "utf8";

    function readbuf() {
        bufread = fs.readSync(fd,buffer,0,bufsize,position);
        position += bufread;
        eof = bufread ? false : true;
        data += buffer.toString(encoding,0,bufread);
    }

    function getLine() {
        var nl = data.indexOf("\r"), hasnl = nl !== -1;
        if (!hasnl && eof) return fs.closeSync(fd), online(data,++lines), onend(lines); 
        if (!hasnl && !eof) readbuf(), nl = data.indexOf("\r"), hasnl = nl !== -1;
        if (!hasnl) return process.nextTick(getLine);
        var line = data.substr(0,nl);
        data = data.substr(nl+1);
        if (data[0] === "\n") data = data.substr(1);
        online(line,++lines);
        process.nextTick(getLine);
    }
    getLine();
}

我遇到了同样的问题并想出了上面的解决方案 其他人看起来很相似，但它是 aSync 并且可以非常快速地读取大文件

希望对你有帮助

Answer 20

我有一个小模块，它做得很好，并被很多其他项目使用 npm readline 请注意，在节点 v10 中有一个本机 readline 模块，所以我将我的模块重新发布为 linebyline https://www.npmjs.com/package/linebyline

如果你不想使用该模块，功能很简单：

var fs = require('fs'),
EventEmitter = require('events').EventEmitter,
util = require('util'),
newlines = [
  13, // \r
  10  // \n
];
var readLine = module.exports = function(file, opts) {
if (!(this instanceof readLine)) return new readLine(file);

EventEmitter.call(this);
opts = opts || {};
var self = this,
  line = [],
  lineCount = 0,
  emit = function(line, count) {
    self.emit('line', new Buffer(line).toString(), count);
  };
  this.input = fs.createReadStream(file);
  this.input.on('open', function(fd) {
    self.emit('open', fd);
  })
  .on('data', function(data) {
   for (var i = 0; i < data.length; i++) {
    if (0 <= newlines.indexOf(data[i])) { // Newline char was found.
      lineCount++;
      if (line.length) emit(line, lineCount);
      line = []; // Empty buffer.
     } else {
      line.push(data[i]); // Buffer new line data.
     }
   }
 }).on('error', function(err) {
   self.emit('error', err);
 }).on('end', function() {
  // Emit last line if anything left over since EOF won't trigger it.
  if (line.length){
     lineCount++;
     emit(line, lineCount);
  }
  self.emit('end');
 }).on('close', function() {
   self.emit('close');
 });
};
util.inherits(readLine, EventEmitter);

Answer 21

另一种解决方案是通过顺序执行器nsynjs 运行逻辑。它使用 node readline 模块逐行读取文件，并且不使用承诺或递归，因此不会在大文件上失败。下面是代码的样子：

var nsynjs = require('nsynjs');
var textFile = require('./wrappers/nodeReadline').textFile; // this file is part of nsynjs

function process(textFile) {

    var fh = new textFile();
    fh.open('path/to/file');
    var s;
    while (typeof(s = fh.readLine(nsynjsCtx).data) != 'undefined')
        console.log(s);
    fh.close();
}

var ctx = nsynjs.run(process,{},textFile,function () {
    console.log('done');
});

以上代码基于此示例：https://github.com/amaksr/nsynjs/blob/master/examples/node-readline/index.js

Answer 22

这是我最喜欢的文件浏览方式，一种简单的本地解决方案，用于使用现代async/await 读取渐进式（不是“slurp”或全内存方式）文件。在处理大型文本文件时，我发现这是一种“自然”的解决方案，无需求助于 readline 包或任何非核心依赖项。

let buf = '';
for await ( const chunk of fs.createReadStream('myfile') ) {
    const lines = buf.concat(chunk).split(/\r?\n/);
    buf = lines.pop();
    for( const line of lines ) {
        console.log(line);
    }
}
if(buf.length) console.log(buf);  // last line, if file does not end with newline

您可以在fs.createReadStream 中调整编码或使用chunk.toString(<arg>)。这也让你更好地根据你的喜好微调分割线，即。使用.split(/\n+/) 跳过空行并使用{ highWaterMark: <chunkSize> } 控制块大小。

不要忘记创建一个类似processLine(line) 的函数，以避免由于结尾buf 剩余而重复行处理代码两次。不幸的是，ReadStream 实例在此设置中没有更新其文件结束标志，因此，afaik 无法在循环中检测到我们处于最后一次迭代中，而不需要一些更详细的技巧，例如比较文件从fs.Stats() 到.bytesRead 的大小。因此最终的buf 处理解决方案，除非您绝对确定您的文件以换行符\n 结尾，在这种情况下for await 循环就足够了。

★ 如果您更喜欢事件异步版本，那就是它：

let buf = '';
fs.createReadStream('myfile')
.on('data', chunk => {
    const lines = buf.concat(chunk).split(/\r?\n/);
    buf = lines.pop();
    for( const line of lines ) {
        console.log(line);
    }
})
.on('end', () => buf.length && console.log(buf) );

★ 现在，如果您不介意导入 stream 核心包，那么这是等效的管道流版本，它允许链接转换，如 gzip 解压缩：

const { Writable } = require('stream');
let buf = '';
fs.createReadStream('myfile').pipe(
    new Writable({
        write: (chunk, enc, next) => {
            const lines = buf.concat(chunk).split(/\r?\n/);
            buf = lines.pop();
            for (const line of lines) {
                console.log(line);
            }
            next();
        }
    })
).on('finish', () => buf.length && console.log(buf) );

Answer 23

在进行此类操作时，我们必须问自己两个问题：

1. 执行它所用的内存量是多少？
2. 内存消耗是否随着文件大小而急剧增加？

require('fs').readFileSync() 之类的解决方案会将整个文件加载到内存中。这意味着执行操作所需的内存量几乎等于文件大小。我们应该避免这些大于50mbs

我们可以通过在函数调用之后放置这些代码行来轻松跟踪函数使用的内存量：

    const used = process.memoryUsage().heapUsed / 1024 / 1024;
    console.log(
      `The script uses approximately ${Math.round(used * 100) / 100} MB`
    );

现在从大文件中读取特定行的最佳方法是使用节点的readline。该文档有惊人的examples。

Answer 24

我用这个：

function emitLines(stream, re){
    re = re && /\n/;
    var buffer = '';

    stream.on('data', stream_data);
    stream.on('end', stream_end);

    function stream_data(data){
        buffer += data;
        flush();
    }//stream_data

    function stream_end(){
        if(buffer) stream.emmit('line', buffer);
    }//stream_end


    function flush(){
        var re = /\n/;
        var match;
        while(match = re.exec(buffer)){
            var index = match.index + match[0].length;
            stream.emit('line', buffer.substring(0, index));
            buffer = buffer.substring(index);
            re.lastIndex = 0;
        }
    }//flush

}//emitLines

在流上使用此函数并监听将发出的行事件。

gr-

Answer 25

虽然您可能应该按照最佳答案的建议使用readline 模块，但readline 似乎面向命令行界面而不是行阅读。关于缓冲，它也有点不透明。 （任何需要面向流线的阅读器的人都可能想要调整缓冲区大小）。 readline 模块大约有 1000 行，而带有统计和测试的这个模块是 34 行。

const EventEmitter = require('events').EventEmitter;
class LineReader extends EventEmitter{
    constructor(f, delim='\n'){
        super();
        this.totalChars = 0;
        this.totalLines = 0;
        this.leftover = '';

        f.on('data', (chunk)=>{
            this.totalChars += chunk.length;
            let lines = chunk.split(delim);
            if (lines.length === 1){
                this.leftover += chunk;
                return;
            }
            lines[0] = this.leftover + lines[0];
            this.leftover = lines[lines.length-1];
            if (this.leftover) lines.pop();
            this.totalLines += lines.length;
            for (let l of lines) this.onLine(l);
        });
        // f.on('error', ()=>{});
        f.on('end', ()=>{console.log('chars', this.totalChars, 'lines', this.totalLines)});
    }
    onLine(l){
        this.emit('line', l);
    }
}
//Command line test
const f = require('fs').createReadStream(process.argv[2], 'utf8');
const delim = process.argv[3];
const lineReader = new LineReader(f, delim);
lineReader.on('line', (line)=> console.log(line));

这是一个更短的版本，没有统计信息，只有 19 行：

class LineReader extends require('events').EventEmitter{
    constructor(f, delim='\n'){
        super();
        this.leftover = '';
        f.on('data', (chunk)=>{
            let lines = chunk.split(delim);
            if (lines.length === 1){
                this.leftover += chunk;
                return;
            }
            lines[0] = this.leftover + lines[0];
            this.leftover = lines[lines.length-1];
            if (this.leftover) 
                lines.pop();
            for (let l of lines)
                this.emit('line', l);
        });
    }
}

Answer 26

const fs = require("fs")

fs.readFile('./file', 'utf-8', (err, data) => {
var innerContent;
    console.log("Asynchronous read: " + data.toString());
    const lines = data.toString().split('\n')
    for (let line of lines)
        innerContent += line + '<br>';


});

Answer 27

我将日常线路处理的整个逻辑封装为一个 npm 模块：line-kit https://www.npmjs.com/package/line-kit

// example
var count = 0
require('line-kit')(require('fs').createReadStream('/etc/issue'),
                    (line) => { count++; },
                    () => {console.log(`seen ${count} lines`)})

Answer 28

在验证它不是目录并且它不包含在不需要检查的文件列表中之后，我使用下面的代码读取行。

(function () {
  var fs = require('fs');
  var glob = require('glob-fs')();
  var path = require('path');
  var result = 0;
  var exclude = ['LICENSE',
    path.join('e2e', 'util', 'db-ca', 'someother-file'),
    path.join('src', 'favicon.ico')];
  var files = [];
  files = glob.readdirSync('**');

  var allFiles = [];

  var patternString = [
    'trade',
    'order',
    'market',
    'securities'
  ];

  files.map((file) => {
    try {
      if (!fs.lstatSync(file).isDirectory() && exclude.indexOf(file) === -1) {
        fs.readFileSync(file).toString().split(/\r?\n/).forEach(function(line){
          patternString.map((pattern) => {
            if (line.indexOf(pattern) !== -1) {
              console.log(file + ' contain `' + pattern + '` in in line "' + line +'";');
              result = 1;
            }
          });
        });
      }
    } catch (e) {
      console.log('Error:', e.stack);
    }
  });
  process.exit(result);

})();

Answer 29

上面的答案我都看过了，都是用第三方库来解决的。它在 Node 的 API 中有一个简单的解决方案。例如

const fs= require('fs')

let stream = fs.createReadStream('<filename>', { autoClose: true })

stream.on('data', chunk => {
    let row = chunk.toString('ascii')
}))