节点：fs write() 不在循环内写入。为什么不？

Question

我想创建一个写入流并在我的数据进入时写入它。但是，我能够创建文件但没有写入任何内容。最终，进程耗尽内存。

我发现问题是我在循环中调用 write()。

这是一个简单的例子：

'use strict'

var fs = require('fs');
var wstream = fs.createWriteStream('myOutput.txt');

for (var i = 0; i < 10000000000; i++) {
    wstream.write(i+'\n');
}

console.log('End!')
wstream.end();

什么都没有写，甚至没有你好。但为什么？如何在循环中写入文件？

Answer 1

为了补充（甚至更多）@Mike Cluck 的answer，我使用节点流pipe() 实现了具有相同行为的解决方案。也许它对某人有用。 根据docs（节点11.13.0）：

readable.pipe() 方法将 Writable 流附加到可读流， 使其自动切换到流动模式并推动所有 其数据到附加的 Writable 中。 数据流将是 自动管理，以便目标可写流不是 被更快的可读流淹没。

所以，pipe() 提供了开箱即用的背压策略。所需要的只是以某种方式创建可读流。在我的示例中，我从节点流模块扩展 Readable 类以创建简单的计数器：

const { Readable } = require('stream');
const fs = require('fs');
const writeStream = fs.createWriteStream('./bigFile.txt');

class Counter extends Readable {
    constructor(opt) {
        super(opt);
        this._max = 1e7;
        this._index = 1;
    }

    _read() {
        const i = this._index++;
        if (i > this._max)
            this.push(null);
        else {
            this.push(i + '\n');
        }
    }
}

new Counter().pipe(writeStream); 

行为完全相同 - 数据以小块不断推送到文件，并且内存消耗是恒定的（在我的机器上约为 50MB）。

pipe() 的优点在于，如果您提供了可读流（即来自请求），您需要做的就是使用：readable.pipe(writable)。

Answer 2

作为对@MikeC 的excellent answer 的补充，以下是writable.write() 的当前文档（v8.4.0）中的一些相关细节：

如果返回false，则应停止向流中写入数据的进一步尝试，直到发出'drain' 事件。

当流没有耗尽时，对write() 的调用将缓冲chunk，并返回false。一旦所有当前缓冲的块被耗尽（操作系统接受交付），'drain' 事件将被发出。建议一旦write() 返回false，在发出'drain' 事件之前不要再写入块。虽然允许在未耗尽的流上调用 write()，但 Node.js 将缓冲所有写入的块，直到达到最大内存使用量，此时它将无条件中止。甚至在它中止之前，高内存使用会导致垃圾收集器性能不佳和高RSS（通常不会释放回系统，即使在不再需要内存之后也是如此）。

对于backpressuring in streams：

在数据缓冲区超过highWaterMark 或写入队列当前繁忙的任何情况下，.write() 将返回false。

当返回 false 值时，背压系统启动。

一旦数据缓冲区被清空，将发出.drain() 事件并恢复传入的数据流。

队列完成后，背压将允许再次发送数据。正在使用的内存空间将自行释放并为下一批数据做准备。

               +-------------------+         +=================+
               |  Writable Stream  +--------->  .write(chunk)  |
               +-------------------+         +=======+=========+
                                                     |
                                  +------------------v---------+
   +-> if (!chunk)                |    Is this chunk too big?  |
   |     emit .end();             |    Is the queue busy?      |
   +-> else                       +-------+----------------+---+
   |     emit .write();                   |                |
   ^                                   +--v---+        +---v---+
   ^-----------------------------------<  No  |        |  Yes  |
                                       +------+        +---v---+
                                                           |
           emit .pause();          +=================+     |
           ^-----------------------+  return false;  <-----+---+
                                   +=================+         |
                                                               |
when queue is empty     +============+                         |
^-----------------------<  Buffering |                         |
|                       |============|                         |
+> emit .drain();       |  ^Buffer^  |                         |
+> emit .resume();      +------------+                         |
                        |  ^Buffer^  |                         |
                        +------------+   add chunk to queue    |
                        |            <---^---------------------<
                        +============+

---

这里有一些可视化（使用--max-old-space-size=512 运行 V8 堆内存大小为 512MB 的脚本）。

此可视化显示i 每 10,000 步的heap memory usage（红色）和增量时间（紫色）（X 轴显示i）：

'use strict'

var fs = require('fs');
var wstream = fs.createWriteStream('myOutput.txt');
var latestTime = (new Date()).getTime();
var currentTime;

for (var i = 0; i < 10000000000; i++) {
    wstream.write(i+'\n');
    if (i % 10000 === 0) {
        currentTime = (new Date()).getTime();
        console.log([  // Output CSV data for visualisation
            i,
            (currentTime - latestTime) / 5,
            process.memoryUsage().heapUsed / (1024 * 1024)
        ].join(','));
        latestTime = currentTime;
    }
}

console.log('End!')
wstream.end();

随着内存使用量接近最大限制 512MB，脚本运行速度越来越慢，直到达到限制时它最终崩溃。

---

此可视化使用v8.setFlagsFromString() 和--trace_gc 来显示每个垃圾回收的当前内存使用情况（红色）和执行时间（紫色）（X 轴以秒为单位显示总经过时间）：

'use strict'

var fs = require('fs');
var v8 = require('v8');
var wstream = fs.createWriteStream('myOutput.txt');

v8.setFlagsFromString('--trace_gc');

for (var i = 0; i < 10000000000; i++) {
    wstream.write(i+'\n');
}

console.log('End!')
wstream.end();

大约 4 秒后内存使用率达到 80%，而垃圾收集器 gives up trying to Scavenge and is forced to use Mark-sweep（慢了 10 倍以上）——请参阅 this article 了解更多详细信息。

---

为了比较，下面是@MikeC 的代码的相同可视化，当write 缓冲区变满时等待drain：

Answer 3

问题是你从来没有给它机会耗尽缓冲区。最终这个缓冲区满了，你的内存就用完了。

WriteStream.write 返回一个布尔值，指示数据是否成功写入磁盘。如果数据没有写入成功，则等待drain event，表示缓冲区已被耗尽。

这是编写代码的一种方法，它利用write 的返回值和drain 事件：

'use strict'

var fs = require('fs');
var wstream = fs.createWriteStream('myOutput.txt');

function writeToStream(i) {
  for (; i < 10000000000; i++) {
    if (!wstream.write(i + '\n')) {
      // Wait for it to drain then start writing data from where we left off
      wstream.once('drain', function() {
        writeToStream(i + 1);
      });
      return;
    }
  }
  console.log('End!')
  wstream.end();
}

writeToStream(0);