如何优化Readable Stream？咨询IncomingMessage更优读取方案

我有一段能正常运行的代码，但关于Readable流的资料大多比较陈旧，想请教几个问题：

• 这段代码有没有遗漏的地方？
• 有没有更优的读取IncomingMessage的方式？
• 我试过用doc.on('data', chunk => {})但没生效，这是为什么？
• 不太理解流是怎么判断读取结束的——仅仅是靠.read()返回null吗？

我用Gotenberg的live demo端点来获取这个流做测试，另外已经把代码包装在Promise中，方便使用async-await语法。

async function generateBufferFromIncomingMessage(doc) {
  return new Promise((resolve, reject) => {
    const chunks = []

    doc.on('readable', (chunk) => {
      // console.log('readable')
      while (null !== (chunk = doc.read())){
        chunks.push(chunk)
      }
    })

    doc.on('end', () => {
      // console.log('end')
      const result = Buffer.concat(chunks)
      resolve(result)
    })

    doc.on('error', (error) => {
      console.log(error)
      reject(error)
    })
  })
}

1. 现有代码的遗漏点 #

你的代码整体功能没问题，但有个小细节：readable事件的回调参数是无效的——该事件触发时不会传递chunk数据，必须通过doc.read()主动读取。另外，错误处理里的console.log(error)可以保留，但生产环境建议换成更规范的日志方案，不过这不会影响核心功能。

2. 更优的读取方式 #

Node.js从v10开始内置了stream.promises模块，用它可以大幅简化流的读取逻辑，不用手动拼接chunk：

const { finished } = require('stream/promises');
const { Buffer } = require('buffer');

// 写法1：for await...of 遍历流
async function generateBufferFromIncomingMessage(doc) {
  const chunks = [];
  for await (const chunk of doc) {
    chunks.push(chunk);
  }
  return Buffer.concat(chunks);
}

// 写法2：结合finished方法监听流结束
async function generateBufferFromIncomingMessage(doc) {
  const chunks = [];
  doc.on('data', chunk => chunks.push(chunk));
  await finished(doc);
  return Buffer.concat(chunks);
}

这两种写法都比手动监听readable更现代，也更贴合async-await的语法风格。

3. data事件没生效的原因 #

大概率是你绑定data事件的时机太晚——如果在绑定前已经调用过doc.read()或其他触发流流动的操作，早期的chunk已经被消费，data事件自然捕获不到。另外，Node.js流默认是暂停模式，仅绑定data事件会自动切换到流动模式，但如果之前用readable模式操作过，需要手动调用doc.resume()启动流动：

async function generateBufferFromIncomingMessage(doc) {
  return new Promise((resolve, reject) => {
    const chunks = [];
    doc.on('data', chunk => chunks.push(chunk));
    doc.on('end', () => resolve(Buffer.concat(chunks)));
    doc.on('error', reject);
    doc.resume(); // 确保流进入流动模式
  });
}

4. 流的结束判断逻辑 #

流的结束判断不是只靠.read()返回null：

• 在暂停模式（用readable事件）下，doc.read()返回null仅表示当前没有可用数据，不代表流结束，只有end事件触发时，才说明所有数据读取完成。
• 在流动模式（用data事件）下，end事件触发就意味着流已结束，不会再有data事件触发。
  你的代码里用while (null !== (chunk = doc.read()))循环读取，是因为readable事件可能多次触发，每次触发时可能有多个chunk可用，直到read()返回null，此时等待下一次readable或end事件即可。

内容的提问来源于stack exchange，提问作者jean d'arme