Node.js可读流pipe至response：遗留音视频流代码outgoingData重复读流问题

这种重复读取流数据的问题在维护旧版Node.js代码时确实很头疼，结合你描述的第三方设备音视频流pipe到response的场景，我来梳理几个大概率的原因和对应的解决办法：

可能的核心原因 #

• 流实例复用问题：如果你的代码中复用了同一个readable stream实例处理多次请求，旧版Node.js的stream在被消费后，内部缓存的未读取数据或者流状态没有被重置，就会导致后续请求重复读取之前的片段。
• 流模式切换异常：Node.js 8的readable stream有flowing和paused两种模式，如果手动监听了data事件但没有正确处理模式切换，或者pipe()的时机不对，可能会导致流的缓存数据被重复触发读取。
• 缺少流销毁/清理逻辑：当请求结束或出现错误时，没有及时销毁readable stream，导致流的内部状态残留，下次请求时继续读取旧数据。

具体解决步骤 #

1. 确保每次请求创建新的Stream实例 #

检查你的outgoingData端点代码，绝对不要复用全局或缓存的stream实例。每次收到请求时，都应该重新从第三方设备获取新的流对象，比如：

// 错误示例：复用全局stream
const globalStream = getThirdPartyStream();
app.get('/outgoingData', (req, res) => {
  globalStream.pipe(res);
});

// 正确示例：每次请求创建新stream
app.get('/outgoingData', (req, res) => {
  const newStream = getThirdPartyStream(); // 每次请求都获取新的流
  newStream.pipe(res);
});

2. 完善Stream的事件监听与清理 #

在Node.js 8中，pipe()虽然会自动处理流的消费，但一定要添加error和finish事件监听，确保流在异常或完成后被正确销毁：

app.get('/outgoingData', (req, res) => {
  const stream = getThirdPartyStream();
  
  stream.pipe(res)
    .on('error', (err) => {
      console.error('Pipe error:', err);
      stream.destroy(); // 销毁流，重置内部状态
      res.status(500).end('Stream error');
    })
    .on('finish', () => {
      console.log('Stream piping completed');
      stream.destroy(); // 完成后销毁流，避免内存泄漏
    });

  // 同时监听response的close事件，避免客户端断开后流继续运行
  res.on('close', () => {
    stream.destroy();
  });
});

3. 检查Stream的读取模式 #

如果你的代码中手动监听了data事件（而不是用pipe()），要确保正确切换流的模式。更推荐直接用pipe()，因为旧版Node.js中pipe()会自动处理流模式和消费，避免手动处理的疏漏：

// 不推荐手动处理data事件（容易出问题）
stream.on('data', (chunk) => {
  res.write(chunk);
});
stream.on('end', () => {
  res.end();
});

// 推荐使用pipe()，自动处理流模式和消费
stream.pipe(res);

4. 排查第三方设备的流输出 #

如果上面的步骤都没解决问题，可能是第三方设备在每次请求时重复发送了相同的初始数据片段。可以在代码中打印chunk的哈希值，对比每次请求的第一个chunk是否一致，以此确认是否是设备端的问题：

stream.on('data', (chunk) => {
  console.log('Chunk hash:', require('crypto').createHash('md5').update(chunk).digest('hex'));
  res.write(chunk);
});

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