将同一个可读流导入多个（可写）目标的Node.js管道

Question

将同一个可读流导入多个（可写）目标的Node.js管道

javascriptnode.jsstreampipenode.js-stream

101

我需要依次运行两个命令，并且这两个命令都需要从同一流中读取数据。在将一个流导入另一个流之后，缓冲区被清空，所以我无法再从该流中读取数据，因此以下操作不起作用：

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 生成的流（或任何其他流）。

是否有一种方法可以在流完成管道传输后重复使用可读流？如何实现类似上面示例的最佳方法呢？

- Maroshii

看起来你正在使用imagemick。你可以像50%这样传递值给-scale进行缩放。你也可以使用https://npmjs.org/package/gm。 - user568109

3

@user568109 好的。然而这不是问题的关键。这是一个更一般的问题... 它是有关imagemagick或任何其他命令/流的。 - Maroshii

7个回答

12

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

以下模式对我非常有效。它使用基于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); });

- artikas

1

哪一部分实际上覆盖了数据？覆盖数据的代码应该自然地抛出一个错误。 - Robert Siemer

5

您可以使用我创建的小型npm包：
readable-stream-clone
使用它，您可以将可读流重复使用多次。

- levansuper

3

它是否遭受了上面所描述的背压问题？第二个管道产生一个空文件呢？如果你能再详细解释一下，那就太好了（对我和你的包名声都有好处 :-)）。先谢谢了！ - maganap

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

还有一种更加智能的替代实现：https://github.com/mcollina/cloneable-readable - SleepWalker

4

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

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

- Yin

1

我有一个不同的解决方案，可以同时写入两个流，自然地，写入时间将是两个时间的总和，但我用它来响应下载请求，在这里，我想在我的服务器上保留已下载文件的副本（实际上我使用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 传递给您的方法。

- Zied Hamdi

在不检查返回值为false（或缓冲区已满）的情况下向流中写入数据会导致内存泄漏，请考虑在Node.js中如何实现背压。https://nodejs.org/en/docs/guides/backpressuring-in-streams - undefined

1

如果您在PassThrough流上有异步操作，这里发布的答案将无法工作。对于异步操作有效的解决方案包括缓冲流内容，然后从缓冲结果创建流。

To buffer the result you can use 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);
    });
}

To create streams from the buffer you can use:

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

- Juan

我觉得这样做违背了使用流的初衷，因为你是把流加载到内存中，而不是随着流的到来进行处理。 - undefined

-1

管道如何分别输入到两个或更多个流中，但不是同时进行？

例如：

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文件为空

- user3683370

在某种程度上，它对我有所帮助！ - sandip

8

我认为你已经找到了一个问题，但是这很令人困惑，因为这不是一个答案。 - Michael

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- user568109 · Accepted Answer

你需要通过将流导入两个流来创建副本。你可以使用一个 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