将NodeJS流消耗到缓冲区并写入流的正确方法

Posted

技术标签:

【中文标题】将NodeJS流消耗到缓冲区并写入流的正确方法【英文标题】:Proper way to consume NodeJS stream into buffer and write stream 【发布时间】:2018-12-24 13:36:49 【问题描述】:

我需要将可读流通过管道传输到缓冲区(要转换为字符串)和文件中。流来自node-fetch

NodeJS 流有两种状态:暂停和流动。据我了解,只要附加了'data' 侦听器,流就会变为流动模式。我想确保读取流的方式不会丢失任何字节。

方法一:管道和从'data'读取:

fetch(url).then(
  response =>
    new Promise(resolve => 
      const buffers = []
      const dest = fs.createWriteStream(filename)
      response.body.pipe(dest)
      response.body.on('data', chunk => buffers.push(chunk))
      dest.on('close', () => resolve(Buffer.concat(buffers).toString())
    )
)

方法2:使用直通流:

const  PassThrough  = require('stream')
fetch(url).then(
  response =>
    new Promise(resolve => 
      const buffers = []
      const dest = fs.createWriteStream(filename)
      const forFile = new PassThrough()
      const forBuffer = new PassThrough()
      response.body.pipe(forFile).pipe(dest)
      response.body.pipe(forBuffer)
      forBuffer.on('data', chunk => buffers.push(chunk))
      dest.on('close', () => resolve(Buffer.concat(buffers).toString())
    )
)

是否需要第二种方法才能不丢失数据?第二种方法是否浪费,因为可以缓冲另外两个流?或者,是否有另一种方法可以同时填充缓冲区和写入流?

【问题讨论】:

它实际上按预期工作,我只是不确定我是否足够幸运拥有快速写入流。我读过一些帖子说 write steams 会尝试以自己的速度调用方法read(拉),我还读到“数据”监听器会导致 steam 不断流动(推) 如果您只想先将整个文件读入内存,则可以使用fs.writeFile()。没有必要.pipe()它。如果您不需要内存中的整个文件,那么.pipe() 效率更高。 【参考方案1】:

您不会错过任何数据,因为.pipe 内部调用src.on('data') 并将任何块写入目标流。

因此,写入您的dest 流的任何块也将发送到您正在缓冲块的response.body.on('data')。 在任何情况下,您都应该监听'error' 事件并在出现任何错误时拒绝。

虽然您的第二种模式可以工作,但您不需要它。


这是来自.pipe 函数的一段代码

  src.on('data', ondata);
  function ondata(chunk) 
    debug('ondata');
    var ret = dest.write(chunk);
    debug('dest.write', ret);
    if (ret === false) 
      // If the user unpiped during `dest.write()`, it is possible
      // to get stuck in a permanently paused state if that write
      // also returned false.
      // => Check whether `dest` is still a piping destination.
      if (((state.pipesCount === 1 && state.pipes === dest) ||
           (state.pipesCount > 1 && state.pipes.indexOf(dest) !== -1)) &&
          !cleanedUp) 
        debug('false write response, pause', state.awaitDrain);
        state.awaitDrain++;
      
      src.pause();
    
  

【讨论】:

以上是关于将NodeJS流消耗到缓冲区并写入流的正确方法的主要内容,如果未能解决你的问题,请参考以下文章

IO流:缓冲流

使用带缓冲区的输入输出流的速度会大幅提高

java中的缓冲流

java字节缓冲流和字符缓冲流

IO流之缓冲流

IO流25 - 字符流 - 字符输出流的缓冲流以及字符输入流的缓冲流BufferedWriter和BufferedReader