Node.js中如何正确清空可读流内部缓冲区 避免重pipe时发送残留数据

Node.js 可读流的内部缓冲队列不会因为unpipe操作自动清空，所有通过push传入但未被消费者消费的数据都会持续暂存在队列中，再次执行pipe操作时队列中所有残留数据都会被一次性发送到下游可写流，这就是你示例中bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb被意外输出的原因。

方案1：临时手动清空缓冲区（适合测试/简单场景） #

你可以直接封装一个工具函数清空可读流的内部缓冲队列，在重pipe前调用即可：

function clearReadableBuffer(readable) {
  const rsState = readable._readableState
  // 兼容Node.js各版本的缓冲队列实现
  rsState.buffer?.clear()
  rsState.length = 0
}

对应你的示例代码，调用位置如下：

console.log('unpipe')
readable.unpipe(writable)
await wait(100)

readable.push('bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb')

clearReadableBuffer(readable) // 重pipe前清空缓冲区

console.log('pipe')
readable.pipe(writable)
await wait(100)

修改后运行代码，bbbb数据块会被直接丢弃，不会输出到下游可写流，符合预期。

方案2：自定义流内置缓冲清理逻辑（适合生产级自定义Transform流） #

如果你正在开发自定义Transform流，建议直接在流内部监听pipe/unpipe事件实现自动清理，不需要依赖外部调用，兼容性和稳定性更好：

const { Transform } = require('stream')

class NoResidueTransform extends Transform {
  constructor(opts) {
    super(opts)
    // 监听pipe事件标记当前处于活跃传输状态
    this.on('pipe', () => {
      this._isActive = true
    })
    // 监听unpipe事件标记非活跃并清空缓冲区
    this.on('unpipe', () => {
      this._isActive = false
      const rsState = this._readableState
      rsState.buffer?.clear()
      rsState.length = 0
    })
  }

  _transform(chunk, encoding, callback) {
    // 非活跃状态下直接丢弃数据，不推入缓冲队列
    if (this._isActive) {
      this.push(chunk)
    }
    callback()
  }
}

使用这个自定义流替代原生的Readable/Transform即可实现重pipe时无残留数据发送的效果。

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