【问题标题】：How do I apply back-pressure to node streams?如何对节点流应用背压？
【发布时间】：2016-03-17 12:56:45
【问题描述】：

在尝试使用 Node.JS 流时，我遇到了一个有趣的难题。当输入（可读）流推送更多数据时，目标（可写）关心我无法正确应用背压。

我尝试的两种方法是从 Writable.prototype._write 返回 false 并保留对 Readable 的引用，以便我可以从 Writable 调用 Readable.pause()。这两种解决方案都没有多大帮助，我将对此进行解释。

在我的练习中（你可以查看the full source as a Gist）我有三个流：

可读 - 密码生成器

util.inherits(PasscodeGenerator, stream.Readable);
function PasscodeGenerator(prefix) {
  stream.Readable.call(this, {objectMode: true});
  this.count  = 0;
  this.prefix = prefix || '';
}
PasscodeGenerator.prototype._read = function() {
  var passcode = '' + this.prefix + this.count;
  if (!this.push({passcode: passcode})) {
    this.pause();
    this.once('drain', this.resume.bind(this));
  }
  this.count++;
};

我认为来自this.push() 的返回码足以自行暂停并等待drain 事件恢复。

变换 - 哈希器

util.inherits(Hasher, stream.Transform);
function Hasher(hashType) {
  stream.Transform.call(this, {objectMode: true});
  this.hashType = hashType;
}
Hasher.prototype._transform = function(sample, encoding, next) {
  var hash = crypto.createHash(this.hashType);
  hash.setEncoding('hex');
  hash.write(sample.passcode);
  hash.end();
  sample.hash = hash.read();
  this.push(sample);
  next();
};

只需将密码的哈希添加到对象。

可写 - SampleConsumer

util.inherits(SampleConsumer, stream.Writable);
function SampleConsumer(max) {
  stream.Writable.call(this, {objectMode: true});
  this.max   = (max != null) ? max : 10;
  this.count = 0;
}
SampleConsumer.prototype._write = function(sample, encoding, next) {
  this.count++;
  console.log('Hash %d (%s): %s', this.count, sample.passcode, sample.hash);
  if (this.count < this.max) {
    next();
  } else {
    return false;
  }
};

在这里，我想尽可能快地使用数据，直到达到最大样本数，然后结束流。我尝试使用 this.end() 而不是 return false 但这导致了可怕的 write call after end 问题。如果样本量很小，则返回 false 会停止一切，但当样本量很大时，我会收到内存不足错误：

FATAL ERROR: CALL_AND_RETRY_LAST Allocation failed - process out of memory
Aborted (core dumped)

根据this SO answer 理论上，Write 流将返回 false 导致流缓冲，直到缓冲区已满（objectMode 默认为 16），最终 Readable 将调用它的 this.pause() 方法。但是16 + 16 + 16 = 48;缓冲区中有 48 个对象，直到东西填满并且系统被阻塞。实际上更少，因为不涉及克隆，因此它们之间传递的对象是相同的引用。那岂不是意味着内存中只有 16 个对象，直到高水位标记停止一切？

最后我意识到我可以让 Writable 引用 Readable 来使用闭包调用它的暂停方法。然而，这个解决方案意味着 Writable 流对另一个对象了解很多。我必须传递一个参考：

var foo = new PasscodeGenerator('foobar');
foo
  .pipe(new Hasher('md5'))
  .pipe(new SampleConsumer(samples, foo));

对于流的工作方式，这感觉不正常。我认为背压足以导致 Writable 阻止 Readable 推送数据并防止内存不足错误。

一个类似的例子是 Unix head 命令。在 Node 中实现这一点，我会假设目标可以结束，而不仅仅是忽略导致源继续推送数据，即使目标有足够的数据来满足文件的开头部分。

我如何惯用地构造自定义流，以便在目标准备结束时源流不会尝试推送更多数据？

【问题讨论】：

标签： node.js stream

【解决方案1】：

这是一个known issue，内部调用_read()。由于您的_read() 总是同步/立即推送，因此内部流实现可以在正确的条件下进入循环。 _read() 实现是 generally expected 执行某种异步 I/O（例如从磁盘或网络读取）。

解决方法（如上面的链接中所述）是使您的_read() 在某些时候至少异步。您也可以在每次调用时使其异步：

PasscodeGenerator.prototype._read = function(n) {
  var passcode = '' + this.prefix + this.count;
  var self = this;

  // `setImmediate()` delays the push until the beginning
  // of the next tick of the event loop
  setImmediate(function() {
    self.push({passcode: passcode});
  });

  this.count++;
};

【讨论】：