Node.js 将相同的可读流传送到多个（可写）目标中

IT技术 javascript node.js stream pipe node.js-stream
2021-02-10 07:43:13

我需要连续运行两个需要从同一个流中读取数据的命令。将流传输到另一个流后，缓冲区被清空，因此我无法再次从该流中读取数据，因此这不起作用：

var spawn = require('child_process').spawn;
var fs = require('fs');
var request = require('request');

var inputStream = request('http://placehold.it/640x360');
var identify = spawn('identify',['-']);

inputStream.pipe(identify.stdin);

var chunks = [];
identify.stdout.on('data',function(chunk) {
  chunks.push(chunk);
});

identify.stdout.on('end',function() {
  var size = getSize(Buffer.concat(chunks)); //width
  var convert = spawn('convert',['-','-scale',size * 0.5,'png:-']);
  inputStream.pipe(convert.stdin);
  convert.stdout.pipe(fs.createWriteStream('half.png'));
});

function getSize(buffer){
  return parseInt(buffer.toString().split(' ')[2].split('x')[0]);
}

请求抱怨这个

Error: You cannot pipe after data has been emitted from the response.

并将inputStream更改fs.createWriteStream为当然会产生相同的问题。我不想写入文件，而是以某种方式重用请求产生的流（或任何其他与此相关的流）。

一旦完成管道，有没有办法重用可读流？完成上述示例的最佳方法是什么？

6个回答

您必须通过管道将流传输到两个流来创建流的副本。您可以使用 PassThrough 流创建一个简单的流，它只是将输入传递给输出。

const spawn = require('child_process').spawn;
const PassThrough = require('stream').PassThrough;

const a = spawn('echo', ['hi user']);
const b = new PassThrough();
const c = new PassThrough();

a.stdout.pipe(b);
a.stdout.pipe(c);

let count = 0;
b.on('data', function (chunk) {
  count += chunk.length;
});
b.on('end', function () {
  console.log(count);
  c.pipe(process.stdout);
});

输出：

8
hi user
将此技术与 Haraka 邮件服务器附件挂钩一起使用，将传入流通过管道传输到多个邮件帐户数据库中。这个答案有效。
2021-03-15 07:43:13
请注意，此技术仅在生成的命令输出未填充背压缓冲区的字节数时才有效。您可以尝试使用 = spawn('head', ['-c', '200K', '/dev/urandom']); 使其失败。如果 c 没有通过管道输出，在某些时候， a.stdout 将暂停管道输出。b 将耗尽并且永无止境。
2021-03-17 07:43:13
不要使用此方法，因为如果以不同的速率读取流，则会产生问题。试试这个npmjs.com/package/readable-stream-clone对我来说效果很好。
2021-03-17 07:43:13
我很困惑，你说你不能处理同一个流两次，但你的解决方案是..处理同一个流两次（使用 PassThrough 转换）。这似乎是矛盾的。这是标准输出流的特别之处吗？
2021-03-22 07:43:13
我对此进行了测试，它确实有效。我认为您说“您不能处理相同的 [the] 流两次”是不正确的，因为这就是您正在做的事情。您关于无法在“结束”后通过管道传输流的第一条陈述是适当的原因。
2021-03-31 07:43:13

第一个答案仅适用于流处理数据所需的时间大致相同的情况。如果花费的时间明显更长，则请求新数据的速度会更快，因此会覆盖速度较慢的数据仍在使用的数据（我在尝试使用重复流解决此问题后遇到了这个问题）。

以下模式对我来说非常有效。它使用基于 Stream2 流、Streamz 和 Promises 的库通过回调同步异步流。使用第一个答案中熟悉的示例：

spawn = require('child_process').spawn;
pass = require('stream').PassThrough;
streamz = require('streamz').PassThrough;
var Promise = require('bluebird');

a = spawn('echo', ['hi user']);
b = new pass;
c = new pass;   

a.stdout.pipe(streamz(combineStreamOperations)); 

function combineStreamOperations(data, next){
  Promise.join(b, c, function(b, c){ //perform n operations on the same data
  next(); //request more
}

count = 0;
b.on('data', function(chunk) { count += chunk.length; });
b.on('end', function() { console.log(count); c.pipe(process.stdout); });
哪个部分实际上覆盖了数据？覆盖的代码自然会抛出错误。
2021-03-15 07:43:13

你可以使用我创建的这个小的 npm 包：

readable-stream-clone

有了这个，您可以根据需要多次重复使用可读流

这个库做正确的事情。很简单，整个源代码都可以复制到这里作为答案。这个库不会受到“背压问题”的影响（参见上面的@maganap 评论）。这个库将完全忽略背压机制。
2021-03-30 07:43:13
@SleepWalker 感谢您的参考
2021-03-31 07:43:13
它遭受描述的背压问题上面？从第二个管道生成一个空文件怎么样？如果你能详细说明一点，那就太棒了（对我和你的包裹声誉:-)）。提前致谢！
2021-04-02 07:43:13
还有更智能的替代实现：github.com/mcollina/cloneable-readable
2021-04-10 07:43:13

对于一般问题，以下代码工作正常

var PassThrough = require('stream').PassThrough
a=PassThrough()
b1=PassThrough()
b2=PassThrough()
a.pipe(b1)
a.pipe(b2)
b1.on('data', function(data) {
  console.log('b1:', data.toString())
})
b2.on('data', function(data) {
  console.log('b2:', data.toString())
})
a.write('text')

我有一个不同的解决方案来同时写入两个流，自然，写入时间将是两次相加，但我用它来响应下载请求，我想在其中保留下载文件的副本我的服务器（实际上我使用的是 S3 备份，所以我在本地缓存了最常用的文件以避免多个文件传输）

/**
 * A utility class made to write to a file while answering a file download request
 */
class TwoOutputStreams {
  constructor(streamOne, streamTwo) {
    this.streamOne = streamOne
    this.streamTwo = streamTwo
  }

  setHeader(header, value) {
    if (this.streamOne.setHeader)
      this.streamOne.setHeader(header, value)
    if (this.streamTwo.setHeader)
      this.streamTwo.setHeader(header, value)
  }

  write(chunk) {
    this.streamOne.write(chunk)
    this.streamTwo.write(chunk)
  }

  end() {
    this.streamOne.end()
    this.streamTwo.end()
  }
}

然后您可以将其用作常规的 OutputStream

const twoStreamsOut = new TwoOutputStreams(fileOut, responseStream)

并将其传递给您的方法，就好像它是响应或 fileOutputStream

其它你可能感兴趣的问题
上一篇在 Babel 6.x 中不能 require() 默认导出值 下一篇动态创建元素的jQuery“on create”事件