通过stdin处理node.js中的海量数据

Question

老问题标题：

"node.js readline form net.Socket (process.stdin) 导致错误：堆内存不足（net.Socket Duplex 到可读流的转换）"

...我已经更改了它，因为没有人回答，这似乎是 node.js 生态系统中的重要问题。

问题是如何解决从巨大的标准输入中逐行读取时“堆内存不足”错误的问题？当您将标准输出转储到文件（例如：test.log）并通过 fs.createReadStream('test.log') 读取到“readline”接口时，不会发生错误。

看起来 process.stdin 不是可读流，因为它在这里提到： https://nodejs.org/api/process.html#process_process_stdin

为了重现该问题，我创建了两个脚本。首先是生成大量数据（a.js 文件）：

// a.js
// loop in this form generates about 7.5G of data
// you can check yourself running:
// node a.js > test.log && ls -lah test.log
// will return
// -rw-r--r--  1 sd  staff   7.5G 31 Jan 22:29 test.log
for (let i = 0 ; i < 8000000 ; i += 1 ) {
  console.log(`${i} ${".".repeat(1000)}\n`);
}

通过带有 readline 的 bash 管道使用它的脚本（b.js 文件）：

const fs = require('fs');

const readline = require('readline');

const rl = readline.createInterface({
  input: process.stdin, // doesn't work
  //input: fs.createReadStream('test.log'), // works
});

let s;

rl.on('line', line => {

  // deliberaty commented out to demonstrate that issue
  // has nothing to do beyond readline and process.stdin

  // s = line.substring(0, 7);
  //
  // if (s === '100 ...' || s === '400 ...' || s === '7500000') {
  //
  //   process.stdout.write(`${line}\n`);
  // }
});

rl.on('error', e => {

  console.log('general error', e)
})

现在当你跑步时；

node a.js | node b.js

会报错：

FATAL ERROR: Ineffective mark-compacts near heap limit Allocation failed - JavaScript heap out of memory

但如果你交换

const rl = readline.createInterface({
  input: process.stdin,
});

到

const rl = readline.createInterface({
  input: fs.createReadStream('test.log')
});

然后运行

node a.js > test.log
node b.js

一切正常

问题实际上归结为如何将 net.Socket 转换为功能齐全的可读流？，如果可能的话。

编辑：

基本上我的问题是，似乎不可能将来自标准输入的大量数据作为流处理，这对于 Unix 风格的管道来说是很自然的。因此，尽管 node.js 在处理流方面非常出色，但您无法编写通过 unix 样式管道处理大量数据的程序。

由于这个限制，在某些情况下完全没有必要将数据转储到硬盘驱动器并且仅在使用 fs.createReadStream('test.log') 处理之后。

我认为流都是关于在飞行中处理大量数据（以及其他用例）而不将其保存在硬盘上。

Answer 1

问题不是输入数据大小，不是节点，而是数据生成器的错误设计：它没有根据消费者输出流的请求实现暂停/恢复数据生成。您应该正确地与标准输出流交互，并正确处理来自该流的pause 和resume 信号，而不仅仅是将数据推送到console.log(..)。

fs.createReadStream() 创建的文件输入流被正确实现，它会根据需要暂停/恢复，因此不会导致代码崩溃。

Answer 2

您始终可以将process.stdin 视为普通的NodeJS 流并处理您自己的读取：

const os = require('os');

function onReadLine(line) {
    // do stuff with line
    console.info(line);
}

// read input and split into lines
let BUFF = '';
process.stdin.on('data', (buff) => {
    const content = buff.toString('utf-8');
    for (let i = 0; i < content.length; i++){
        if (content[i] === os.EOL) {
            onReadLine(BUFF);
            BUFF = '';
        } else {
            BUFF += content[i];
        }
    }
});

// flush last line
process.stdin.on('end', () => {
    if (BUFF.length > 0) {
        onReadLine(BUFF);
    }
});

例子：

// unix
cat ./somefile.txt | node ./script.js

// windows
Start-Process -FilePath "node" -ArgumentList @(".\script.js")  -RedirectStandardInput .\somefile.txt -NoNewWindow -Wait

Answer 3

实际上，这是可能的，您只需将输入流通过管道传输到 Writable 或 Transform Stream。

process.stdin.pipe(YourWritableOrTransformStream);