使用pipe的Node ReadableStream是否需要实现_read()方法？

我从Node Stream教程中获取了以下示例代码并尝试运行：

const Stream = require('stream')

const readableStream = new Stream.Readable()
const writableStream = new Stream.Writable()

writableStream._write = (chunk, encoding, next) => {
    console.log(chunk.toString())
    next()
}

readableStream.pipe(writableStream)

readableStream.push('ping!')
readableStream.push('pong!')

writableStream.end()

运行时出现错误：NodeError: The _read() method is not implemented。我原以为通过pipe输出的readableStream无需实现_read()方法，对此感到困惑。

补充：有观点称“readableStream的_read()方法是推送数据的必填项”，但我看到另一示例代码无需实现_read()方法即可正常运行：

const { Readable } = require("stream");

const inStream = new Readable();

inStream.push("ABCDEFGHIJKLM");
inStream.push("NOPQRSTUVWXYZ");

inStream.push(null); // No more data

inStream.pipe(process.stdout);

问题原因分析 #

这两个示例的核心区别在于是否在调用pipe()前就完成了所有数据的推送并标记流结束：

1. 第一个示例报错的原因：

   • 调用readableStream.pipe(writableStream)之后才推送数据，pipe()启动时，Node.js的流机制会尝试调用_read()方法来触发数据读取逻辑，但该方法未实现，因此抛出错误。
   • 虽然手动调用了push()，但pipe()触发的流状态检查会优先验证_read()是否存在，导致报错。

2. 第二个示例正常运行的原因：

   • 在调用inStream.pipe(process.stdout)前，已经完成所有数据推送，并且通过push(null)标记了可读流结束。
   • 当pipe()启动时，可读流已经处于ended状态，Node.js不会再尝试调用_read()，直接将已有的数据写入目标流，因此不会报错。

第一个示例的两种解决方法 #

• 方法一：提前完成数据推送并标记流结束
  调整代码顺序，在pipe()前推送所有数据并标记可读流结束：

  const Stream = require('stream')
  
  const readableStream = new Stream.Readable()
  const writableStream = new Stream.Writable()
  
  writableStream._write = (chunk, encoding, next) => {
      console.log(chunk.toString())
      next()
  }
  
  // 先推送数据并标记可读流结束
  readableStream.push('ping!')
  readableStream.push('pong!')
  readableStream.push(null)
  
  readableStream.pipe(writableStream)
  

• 方法二：实现空的_read()方法
  即使不需要用它来触发数据推送，也可以实现一个空的_read()方法满足流机制的要求：

  const Stream = require('stream')
  
  const readableStream = new Stream.Readable()
  const writableStream = new Stream.Writable()
  
  // 实现空的_read()方法
  readableStream._read = () => {}
  
  writableStream._write = (chunk, encoding, next) => {
      console.log(chunk.toString())
      next()
  }
  
  readableStream.pipe(writableStream)
  
  readableStream.push('ping!')
  readableStream.push('pong!')
  readableStream.push(null) // 标记可读流结束，管道会自动完成后续流程
  

注意：无需手动调用writableStream.end()，通过给可读流推送null来触发整个管道的结束流程，是更符合Node.js流规范的做法。

内容的提问来源于stack exchange，提问作者xpt