Node.js按需读取Stream问题：自定义异步读取函数未达预期

现有代码的问题 #

• stream.read()是同步方法，用await毫无意义，根本不会等待数据就绪就直接执行，这是代码不达预期的核心原因。
• 用stream.closed判断流状态是错误的，Node.js标准流用readableEnded属性标识流是否已结束。
• 无限while(1)轮询会导致CPU空转，没有利用事件机制等待数据就绪，完全浪费资源。
• end事件里调用stream.close()多余，文件读取流结束后会自动关闭，手动调用反而可能引发异常。

正确的实现方式 #

我们需要用Promise封装流的事件监听，等待readable或end事件触发后再读取数据，这样就能实现按需读取的逻辑：

const fs = require('fs')

// 封装按需读取数据块的Promise方法
const requestChunkFrom = (stream, size) => {
  return new Promise((resolve) => {
    // 先尝试读取一次，可能缓存里已有数据
    const tryRead = () => {
      const data = stream.read(size)
      if (data) {
        return resolve({ chunk: data, isEnd: false })
      }
      // 如果没读到数据，监听事件
      const onReadable = () => {
        cleanupListeners()
        tryRead()
      }
      const onEnd = () => {
        cleanupListeners()
        resolve({ chunk: null, isEnd: true })
      }
      // 清理监听函数，避免内存泄漏
      const cleanupListeners = () => {
        stream.off('readable', onReadable)
        stream.off('end', onEnd)
      }
      stream.on('readable', onReadable)
      stream.on('end', onEnd)
    }
    tryRead()
  })
}

const job = async () => {
  const stream = fs.createReadStream('README.md')
  
  stream.on('end', () => {
    console.log('流已结束')
  })
  stream.on('readable', () => {
    console.log('数据就绪，可以读取')
  })

  let isEnd = false
  do {
    const { chunk, isEnd: chunkEnd } = await requestChunkFrom(stream, 16)
    isEnd = chunkEnd
    console.log(chunk ? chunk.toString() : chunk)
  } while (!isEnd)
}

job()

代码说明 #

• 用Promise包装读取逻辑，只有当readable事件触发（数据就绪）或end事件触发（流结束）时才会resolve结果，避免空轮询。
• 先尝试同步读取，因为流可能已经缓存了数据，不用等待事件。
• 每次读取后清理事件监听，防止内存泄漏。
• 通过监听end事件来准确判断流是否结束，比单纯依赖属性更可靠。

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