node.js - 当通过 es.map() 和 JSONStream.stringify() 将 JSONStream.parsed() 数据传输到文件流时， Node 堆耗尽

Question

我正在尝试通过 JSONStream.parse() 将输入流(从巨大的 GeoJSON 文件创建)通过管道传输以将流分解为对象，然后通过 event-stream.map() 允许我转换对象，然后通过 JSONStream.stringify() 从中创建一个字符串，最后是一个可写的输出流。随着进程的运行，我可以看到 Node 的内存占用量继续增长，直到最终耗尽堆。这是重现问题的最简单的脚本 (test.js):

const fs = require("fs")
const es = require("event-stream")
const js = require("JSONStream")

out = fs.createWriteStream("/dev/null")
process.stdin
    .pipe(js.parse("features.*"))
    .pipe(es.map( function(data, cb) { 
        cb(null, data);
        return;
    } ))
    .pipe(js.stringify("{\n\"type\": \"FeatureCollection\", \"features\": [\n\t", ",\n\t", "\n]\n}"))
    .pipe(out)

一个小的 bash 脚本 (barf.sh) 将无穷无尽的 JSON 流注入(inject) Node 的 process.stdin 将导致 Node 的堆逐渐增长:

#!/bin/bash

echo '{"type":"FeatureCollection","features":['
while :
do
    echo '{"type":"Feature","properties":{"name":"A Street"}, "geometry":{"type":"LineString"} },'
done

这样运行:

barf.sh | node test.js

有几个奇怪的方法可以回避这个问题:

• 删除 fs.createWriteStream() 并将最后一个管道阶段从“.pipe(out)”更改为“.pipe(process.stdout)”，然后将 Node 的标准输出管道传输到/dev/null
• 将异步 es.map() 改为同步 es.mapSync()

前面两个操作中的任何一个都将允许脚本永远运行， Node 的内存占用量低且不变。我在运行 Ubuntu 16.04、8GB RAM 的八核机器上使用 Node v6.3.1、事件流 v3.3.4 和 JSONStream 1.1.4。

我希望有人能帮助我纠正我确定是我的明显错误。

Answer 1

JSONStream 不是streams2 流，所以不支持背压控制。 (关于 streams2 here 有一个简短的总结。)

这意味着数据将在 data 事件中从 parse 流中出来，并且无论消费流是否为他们准备好了。如果管道中某处的读写速度存在差异，就会出现缓冲 - 这就是您所看到的。

您的 barf.sh 工具可以看到通过 stdin 注入(inject)的功能。相反，如果您正在读取一个大文件，您应该能够通过暂停文件的读取流来管理流程。因此，如果您要在 map 回调中插入一些 pause/resume 逻辑，您应该能够让它处理大量文件；只是需要更长的时间。我会尝试这样的事情:

let in = fs.createReadStream("/some/massive/file");
let out = fs.createWriteStream("/dev/null");
in
    .pipe(js.parse("features.*"))
    .pipe(es.map(function(data, cb) {
        // This is just an example; a 10-millisecond wait per feature would be very slow.
        if (!in.isPaused()) {
            in.pause();
            global.setTimeout(function () { in.resume(); }, 10);
        }
        cb(null, data);
        return;
    }))
    .pipe(js.stringify("{\n\"type\": \"FeatureCollection\", \"features\": [\n\t", ",\n\t", "\n]\n}"))
    .pipe(out);

顺便说一句，使用 mapSync 在我的电脑(又旧又慢)上几乎没有什么不同。但是，除非您要在 map 中执行一些异步操作，否则我会选择 mapSync。