将 NodeJS 流消费到缓冲区并写入流的正确方法
Proper way to consume NodeJS stream into buffer and write stream
我需要将可读流通过管道传输到缓冲区(要转换为字符串)和文件中。流来自 node-fetch
.
NodeJS 流有两种状态:暂停和流动。据我了解,一旦附加了 'data'
侦听器,流就会变为流动模式。我想确保我读取流的方式不会丢失任何字节。
方法 1:管道和读取 '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())
})
)
是否需要第二种方法才不会丢失数据?第二种方法是否浪费,因为可以缓冲更多的流?或者,是否有另一种方法可以同时填充缓冲区和写入流?
您不会遗漏任何数据,因为 .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();
}
}
我需要将可读流通过管道传输到缓冲区(要转换为字符串)和文件中。流来自 node-fetch
.
NodeJS 流有两种状态:暂停和流动。据我了解,一旦附加了 'data'
侦听器,流就会变为流动模式。我想确保我读取流的方式不会丢失任何字节。
方法 1:管道和读取 '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())
})
)
是否需要第二种方法才不会丢失数据?第二种方法是否浪费,因为可以缓冲更多的流?或者,是否有另一种方法可以同时填充缓冲区和写入流?
您不会遗漏任何数据,因为 .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();
}
}