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

Posted 2023-02-14

【中文标题】Node.js 将相同的可读流输送到多个（可写）目标中

【英文标题】：Node.js Piping the same readable stream into multiple (writable) targets

【发布时间】：2013-11-02 10:42:57

【问题描述】：

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

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 当然会产生相同的问题。 我不想写入文件，但以某种方式重用 request 产生的流（或任何其他方式）。

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

【问题讨论】：

看来您正在使用 imagemick。您可以将 50% 之类的值传递给 -scale 以进行缩放。你也可以使用npmjs.org/package/gm

@user568109 是的。但这不是这里的问题。这是一个更普遍的问题......它是 imagemagick，因为它可以是任何其他命令/流

【参考方案1】：

您必须通过将其连接到两个流来创建流的副本。您可以使用 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 邮件服务器附件挂钩将传入流通过管道传输到多个邮件帐户数据库。这个答案有效。

请注意，此技术仅在衍生命令输出的字节数未填满背压缓冲区时才有效。您可以尝试使用 a = spawn('head', ['-c', '200K', '/dev/urandom']); 使其失败。如果 c 未通过管道输出，则 a.stdout 将暂停输出。 b 将耗尽并且永远不会结束。

我很困惑，你说你不能两次处理相同的流，但你的解决方案是......处理相同的流两次（使用 PassThrough 转换）。这似乎是矛盾的。标准输出流有什么特别之处吗？

我对此进行了测试，它确实有效。我认为您说“您不能两次处理相同的 [the] 流”是不正确的，因为这就是您正在做的事情。您关于在“结束”之后无法通过管道传输流的第一个陈述是适当的原因。

不要使用此方法，因为如果以不同的速率读取流，它会产生问题。试试这个 npmjs.com/package/readable-stream-clone 对我来说效果很好。

【参考方案2】：

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

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')

【参考方案3】：

第一个答案仅在流需要大致相同的时间来处理数据时才有效。如果需要更长的时间，则较快的将请求新数据，从而覆盖较慢的仍在使用的数据（我在尝试使用重复流解决此问题后遇到了这个问题）。

以下模式对我来说效果很好。它使用基于 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); );

【讨论】：

哪个部分实际上覆盖了数据？覆盖的代码自然会抛出错误。

【参考方案4】：

如果不是同时连接到两个或多个流中呢？

例如：

var PassThrough = require('stream').PassThrough;
var mybiraryStream = stream.start(); //never ending audio stream
var file1 = fs.createWriteStream('file1.wav',encoding:'binary')
var file2 = fs.createWriteStream('file2.wav',encoding:'binary')
var mypass = PassThrough
mybinaryStream.pipe(mypass)
mypass.pipe(file1)
setTimeout(function()
   mypass.pipe(file2);
,2000)

上面的代码没有产生任何错误但是file2是空的

【讨论】：

在某种程度上对我有帮助！

我认为您已经发现了一个问题，但令人困惑，因为这不是答案。

【参考方案5】：

我有一个不同的解决方案同时写入两个流，当然，写入的时间将是两次的相加，但是我用它来响应下载请求，我想保留一份副本下载的文件在我的服务器上（其实我用的是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()
  

然后您可以将其用作常规输出流

const twoStreamsOut = new TwoOutputStreams(fileOut, responseStream)

并将其传递给您的方法，就好像它是响应或文件输出流一样

【参考方案6】：

如果您对 PassThrough 流进行异步操作，则此处发布的答案将不起作用。 适用于异步操作的解决方案包括缓冲流内容，然后从缓冲的结果创建流。

要缓冲结果，您可以使用concat-stream

const Promise = require('bluebird');
const concat = require('concat-stream');
const getBuffer = function(stream)
    return new Promise(function(resolve, reject)
        var gotBuffer = function(buffer)
            resolve(buffer);
        
        var concatStream = concat(gotBuffer);
        stream.on('error', reject);
        stream.pipe(concatStream);
    );

要从缓冲区创建流，您可以使用：

const  Readable  = require('stream');
const getBufferStream = function(buffer)
    const stream = new Readable();
    stream.push(buffer);
    stream.push(null);
    return Promise.resolve(stream);

【参考方案7】：

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

readable-stream-clone

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

【讨论】：

它是否遭受above 描述的背压问题？从第二个管道生成empty file 怎么样？如果您能详细说明一下，那就太棒了（对我和您的包裹声誉而言:-)）。提前致谢！

这个库是正确的。它非常简单，可以将整个源代码复制到这里作为答案。这个库不会遭受“背压问题”（参见上面的@maganap 评论）。这个库将完全忽略背压机制。

还有更智能的替代实现：github.com/mcollina/cloneable-readable

@SleepWalker 感谢您的参考