Node.js流事件循环迭代刷新：多管道内容传递确定性问询

你的观察完全正确，我们来拆解这个问题背后的Node.js流机制：

1. 关于已存在内容的管道传递 #

是的，流中已写入的缓存内容，只有当多个pipe()调用在同一事件循环迭代中执行时，才会被所有下游流接收到。

原因在于Node.js可读流的工作方式：当你调用pipe()时，下游的可写流会触发可读流的resume()（如果它之前处于暂停状态），并立即尝试读取可读流缓存中的所有数据。

在你的第一个可正常运行的例子里：

const { PassThrough } = require('stream')
const inputStream = new PassThrough()
inputStream.write('foo') // 此时流中已存在内容
inputStream.pipe(process.stdout)
inputStream.pipe(process.stderr) // 会在stdout和stderr均输出'foofoo'

两个pipe()都是同步执行的——在process.stdout开始读取缓存之前，process.stderr已经完成了管道连接。所以当inputStream的缓存被读取时，两个下游流都能拿到'foo'（本质上是inputStream把数据分别推给了两个下游）。

而第二个无法正常运行的例子里：

const { PassThrough } = require('stream')
const inputStream = new PassThrough()
inputStream.write('foo')
inputStream.pipe(process.stdout)
setImmediate(() => { inputStream.pipe(process.stderr) }) // 仅输出一次'foo'

• 同步代码执行阶段：inputStream.pipe(process.stdout)触发process.stdout读取inputStream的缓存，把'foo'全部消费掉，此时inputStream的缓存为空。
• 等到setImmediate回调执行（进入下一个事件循环迭代），inputStream.pipe(process.stderr)才执行。这时inputStream已经没有缓存数据了，所以process.stderr只能拿到后续写入流的内容（如果有的话）。

2. 这个场景是确定性的，而非非确定 #

你不用担心非确定性——Node.js的事件循环执行顺序是严格定义的：

1. 先执行所有同步代码
2. 再处理微任务队列（比如Promise回调）
3. 然后处理setImmediate队列
4. 接着处理I/O回调等

在你的第二个例子中，process.stdout一定会在setImmediate回调执行前读完缓存数据，所以结果是100%确定的——只会输出一次'foo'。

解决方案：异步添加管道并获取历史数据 #

如果需要在异步时机添加下游流，同时让它拿到之前已经写入的内容，可以用以下几种方式：

方法1：先连接所有管道，再写入数据 #

这是最简单的方式，确保所有下游都在数据写入前完成管道连接：

const { PassThrough } = require('stream')
const inputStream = new PassThrough()
inputStream.pipe(process.stdout)
setImmediate(() => { 
  inputStream.pipe(process.stderr)
  inputStream.write('foo') // 两个流都会输出'foo'
})

方法2：实现一个"tee"流来复制数据 #

创建一个中间流，把输入的数据复制到所有下游流，这样即使异步添加下游，也能拿到后续数据（如果要拿到历史数据，需要额外缓存历史数据）：

const { PassThrough } = require('stream')

// 实现一个简单的tee工具
function createTee() {
  const downstreamStreams = []
  const teeStream = new PassThrough({
    write(chunk, encoding, callback) {
      // 把数据复制给所有下游
      downstreamStreams.forEach(stream => stream.write(chunk, encoding))
      callback()
    },
    end(callback) {
      downstreamStreams.forEach(stream => stream.end())
      callback()
    }
  })

  return {
    addStream(stream) {
      downstreamStreams.push(stream)
    },
    stream: teeStream
  }
}

// 使用示例
const inputStream = new PassThrough()
const myTee = createTee()

// 先连接第一个下游
myTee.addStream(process.stdout)
inputStream.pipe(myTee.stream)

// 写入数据
inputStream.write('foo')

// 异步添加第二个下游
setImmediate(() => {
  myTee.addStream(process.stderr)
  inputStream.write('bar') // 两个流都会输出'bar'
  // 注意：如果要让新下游拿到之前的'foo'，需要在addStream时手动写入缓存的历史数据
})

方法3：缓存历史数据 #

如果必须在数据写入后添加下游，可以提前缓存所有写入的数据，当新下游连接时，把缓存的数据写入新流：

const { PassThrough } = require('stream')

const inputStream = new PassThrough()
const cachedChunks = []

// 监听data事件缓存数据
inputStream.on('data', chunk => cachedChunks.push(chunk))

inputStream.write('foo')
inputStream.pipe(process.stdout)

setImmediate(() => {
  const stderrStream = inputStream.pipe(process.stderr)
  // 把缓存的数据写入新下游
  cachedChunks.forEach(chunk => stderrStream.write(chunk))
})

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