节点流导致大量内存占用或泄漏

Question

我正在使用节点v0.12.7并希望直接从数据库流到客户端(用于文件下载).但是,在使用流时,我注意到大量内存占用(以及可能的内存泄漏).

使用express,我创建了一个端点,只需将可读流管道到响应中,如下所示:

app.post('/query/stream', function(req, res) {

  res.setHeader('Content-Type', 'application/octet-stream');
  res.setHeader('Content-Disposition', 'attachment; filename="blah.txt"');

  //...retrieve stream from somewhere...
  // stream is a readable stream in object mode

  stream
    .pipe(json_to_csv_transform_stream) // I've removed this and see the same behavior
    .pipe(res);
});

在生产中,可stream读取数据库中的数据.数据量非常大(1M +行).我用虚拟流交换了这个可读流(参见下面的代码)以简化调试并注意到相同的行为:我的内存使用量每次都会跳跃大约200M.有时垃圾收集会启动并且内存会下降一点,但它会线性上升,直到我的服务器内存不足.

我开始使用流的原因是不必将大量数据加载到内存中.这种行为有望吗？

我还注意到,在流式传输时,我的CPU使用率跳跃到100%并阻塞(这意味着其他请求无法处理).

我错误地使用了这个吗？

虚拟可读流代码

// Setup a custom readable
var Readable = require('stream').Readable;

function Counter(opt) {
  Readable.call(this, opt);
  this._max = 1000000; // Maximum number of records to generate
  this._index = 1;
}
require('util').inherits(Counter, Readable);

// Override internal read
// Send dummy objects until max is reached
Counter.prototype._read = function() {
  var i = this._index++;
  if (i > this._max) {
    this.push(null);
  }
  else {
    this.push({
      foo: i,
      bar: i * 10,
      hey: 'dfjasiooas' + i,
      dude: 'd9h9adn-09asd-09nas-0da' + i
    });
  }
};

// Create the readable stream
var counter = new Counter({objectMode: true});

//...return it to calling endpoint handler...

更新

只是一个小小的更新,我从未找到原因.我最初的解决方案是使用集群来生成新进程,以便仍然可以处理其他请求.

我已经更新到节点v4.虽然在处理过程中cpu/mem的使用率仍然很高,但它似乎修复了泄漏(意味着内存使用率下降).

Answer 1

看起来你正在做的一切正确.我复制了您的测试用例,并在v4.0.0中遇到了同样的问题.将它从objectMode中取出并JSON.stringify在你的对象上使用似乎可以防止高内存和高CPU.这导致我内置JSON.stringify似乎是问题的根源.使用流式库JSONStream而不是v8方法为我修复了这个问题.它可以像这样使用:.pipe(JSONStream.stringify()).

Answer 2

更新2:以下是各种Stream API的历史记录:

https://medium.com/the-node-js-collection/a-brief-history-of-node-streams-pt-2-bcb6b1fd7468

0.12使用Streams 3.

更新:对于旧的node.js流,这个答案是正确的.如果可写流无法跟上,则新流API具有暂停可读流的机制.

背压

看起来你已经遇到了经典的"背压"node.js问题. 本文详细解释了它.

但这是一个TL; DR:

你是对的,流用于不必将大量数据加载到内存中.

但不幸的是,流没有一种机制来知道是否可以继续流式传输.溪流是愚蠢的.他们只是尽可能快地在下一个流中投放数据.

在您的示例中,您正在读取大型csv文件并将其流式传输到客户端.问题是,读取文件的速度大于通过网络上传文件的速度.因此,数据需要存储在某个地方,直到它们被成功遗忘为止.这就是为什么在客户端下载完成之前你的内存不断增长的原因.

解决方案是将读取流量调节到管道中最慢流的速度.也就是说,你用另一个流来预读你的阅读流,这将告诉你的阅读流什么时候可以读取下一个数据块.