在 Node.js 中暂停 readline

Question

考虑下面的代码......我试图在阅读前 5 行后暂停流：

var fs          = require('fs');
var readline    = require('readline');
var stream      = require('stream');
var numlines    = 0;
var instream    = fs.createReadStream("myfile.json");
var outstream   = new stream;
var readStream = readline.createInterface(instream, outstream);
readStream.on('line', function(line){
  numlines++;
  console.log("Read " + numlines + " lines");
  if (numlines >= 5) {
    console.log("Pausing stream");
    readStream.pause();
  }
});

输出（下一个复制）表明它在暂停后继续读取行。也许 readline 已经在缓冲区中排队了几行，并且无论如何都将它们提供给我......如果它继续在后台异步读取，这将是有意义的，但根据文档，我不知道是什么应有的行为。关于如何达到预期效果有什么建议吗？

Read 1 lines
Read 2 lines
Read 3 lines
Read 4 lines
Read 5 lines
Pausing stream
Read 6 lines
Pausing stream
Read 7 lines

Answer 1

因此，即使在 pause() 之后，readline 流也倾向于“滴落”（即泄漏一些额外的行）。文档没有说明这一点，但确实如此。

如果您希望 pause() 切换立即出现，您必须创建自己的行缓冲区并自己累积剩余的行。

Answer 2

有点不直观，the pause methods does not stop queued up line events：

调用rl.pause() 不会立即暂停readline.Interface 实例发出的其他事件（包括'line'）。

但是，有一个名为 line-by-line 的第 3 方模块，其中 pause 会暂停 line 事件，直到它恢复为止。

var LineByLineReader = require('line-by-line'),
    lr = new LineByLineReader('big_file.txt');

lr.on('error', function (err) {
  // 'err' contains error object
});

lr.on('line', function (line) {
  // pause emitting of lines...
  lr.pause();

  // ...do your asynchronous line processing..
  setTimeout(function () {

      // ...and continue emitting lines.
      lr.resume();
  }, 100);
});

lr.on('end', function () {
  // All lines are read, file is closed now.
});

（我与该模块无关，只是发现它对处理这个问题很有用。）

Answer 3

补充几点：

.on('pause', function() {
    console.log(numlines)
})

你会得到5。它在node.js document中提到：

• 输入流未暂停并且接收 SIGCONT 事件。 （参见事件 SIGTSTP 和 SIGCONT）

所以，我在线路事件中创建了一个 tmp 缓冲区。使用标志来判断是否触发暂停。

.on('line', function(line) {
   if (paused) {
      putLineInBulkTmp(line);
   } else {
      putLineInBulk(line);
   }
}

然后在 on pause 和 resume：

.on('pause', function() {
    paused = true;
    doSomething(bulk, function(resp) {
        // clean up bulk for the next.
        bulk = [];
        // clone tmp buffer.
        bulk = clone(bulktmp);
        bulktmp = [];
        lr.resume();
    });
})
.on('resume', () => {
  paused = false;
})

用这种方式来处理这种情况。