在 NodeJS 中使用流和异步读取和处理大小文件

Question

我无法逐行处理文件列表。这是我正在使用的代码：

var LineReader = require("line-by-line");
var async = require("async");
var files = [ "small.txt", "medium.txt", "large.txt" ];

var queue = async.queue(function(task, next){ console.log(task); next(); }, 10);

async.eachSeries(
    files,
    function (file, callback) {
        var lineReader = new LineReader(file, { encoding: "utf8", skipEmptyLines: true });

        lineReader.on("error", function (err) {
            callback(err);
        });

        lineReader.on("line", function (line) {
            lineReader.pause();
            queue.push(line);
        });

        queue.drain = function () {
            lineReader.resume(); // I need to resume the stream !
            callback(); // When all lines have been processed, I need to read the next file
        };
    },
    function (err) {
        if (err) return console.log(err);
        console.log("Job done.");
    }
);

我使用async“同步”处理每个文件并处理队列中的每一行，而line-by-line逐行读取每个文件。

我的问题是：

• 如果我暂停流，将行推入队列并在我收到此错误后恢复流

RangeError: 超出最大调用堆栈大小

• 如果我暂停流，将行推入队列并等待队列为空，我无法恢复流并执行回调

q.drain = function () { lineReader.resume();打回来（）; };

我怎样才能等到所有行都被处理并执行回调来处理下一个文件？

谢谢。

更新：

我发现“逐行”模块很奇怪。 “结束”事件被发出两次。所以我决定重构代码，我发现问题出在哪里。另一个问题：模块一年没有更新，1个月前发送了2个拉取请求。

这是我的解决方案（如果逐行有效）：

var LineReader = require("line-by-line");
var async = require("async");
var files = [ "small.txt", "medium.txt", "large.txt" ];

var queue = async.queue(function(task, next){ console.log(task); next(); }, 10);

async.eachSeries(
    files,
    function (file, callback) {
        var lineReader = new LineReader(file, { encoding: "utf8", skipEmptyLines: true });

        lineReader.on("error", function (err) {
            callback(err);
        });

        lineReader.on("end", function () {
            callback();
        });

        lineReader.on("line", function (line) {
            lineReader.pause();
            queue.push(line);
        });

        queue.drain = function () {
            lineReader.resume();
        };
    },
    function (err) {
        if (err) return console.log(err);
        console.log("Job done.");
    }
);

使用此解决方案，队列中只有 1 行。如果有人有想法推送超过 1 行然后暂停流。

我会尝试寻找另一个没有这个问题的模块，因为我不想为此重写一个新模块。

Answer 1

我会以完全不同的方式解决这个问题。

使用新的stream API，无需监听事件或暂停。
我会像这样使用gulp 和through2：

var gulp = require('gulp')
, thr = require('through2').obj
;

function fixLine (line) {
  // do stuff with a single line of a file.
  // just return it back for no reason :)
  return line
}

files = [ "small.txt", "medium.txt", "large.txt" ]
gulp.src(files).pipe(thr(function(vfs, enc, next){
  // vfs - vinyl filesystem.
  var str = vfs.contents.toString().split('\n').map(fixLine).join('\n')
  vfs.contents = new Buffer(str)
  next(null, vfs)
}))

但是这是异步的。不能保证文件的顺序是数组中的顺序。但显然，这条线是按顺序处理的。

我希望这会有所帮助。

Answer 2

我喜欢用这个功能：

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();
    }

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

    function flush() {
        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;
        }
    }

}

在流上调用此函数时，您的流将开始广播“线”事件\o/