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

问题描述 投票：0回答：8

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

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

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

javascript node.js stream pipe node.js-stream
8个回答
100
投票

您必须通过将流传输到两个流来创建流的副本。您可以使用 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
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); });
7
投票

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

readable-stream-clone

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

5
投票

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

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')
2
投票

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

  1. 要缓冲结果，您可以使用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);
    });
}

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

    const { Readable } = require('stream');
const getBufferStream = function(buffer){
    const stream = new Readable();
    stream.push(buffer);
    stream.push(null);
    return Promise.resolve(stream);
}
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()
  }
}

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

const twoStreamsOut = new TwoOutputStreams(fileOut, responseStream)

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

0
投票

我遇到了同样的问题并找到了解决方案。这是我的实现：

import { Readable } from 'node:stream';

function cloneReadableStream(stream: Readable): [clone: Readable, origin: Readable] {
  /** @type {import('node:stream').ReadableOptions} */
  const options = {
    read() {},
  };

  const origin = new Readable(options);
  const clone = new Readable(options);

  stream
    .on('error', (err) => {
      origin.destroy(err);
      clone.destroy(err);
    })
    // This starts reading data from the original stream.
    // So we should return two new streams from which reading of data has not yet started.
    .on('data', (chunk) => {
      origin.push(chunk);
      clone.push(chunk);
    })
    .on('end', () => {
      origin.push(null);
      clone.push(null);
    });

  return [clone, origin];
}

有一些使用特点：

let stream = new Readable();
const [clone, origin] = cloneReadableStream(stream);

// Should be replaced by the returning stream
stream = origin;

stream.on('data', (chunk) => { ... })
// or stream.pipe(...);
// ^ the same is available for clone

一些解释：

  • 一旦您订阅了流上的 
    data
    事件，您就开始从中提取数据。您只能提取该数据一次，之后数据就会从流中永远消失。
  • 一旦原始流被完全读取，就完成了。您将无法再从中读取内容。
  • 因此我们应该返回两个流，我们将从原始流中传递相同的数据。
  • 您应该使用返回的流之一而不是原始流（原始流将已被读取并将其所有数据传输到返回的流），第二个将充当克隆。

希望这对某人有帮助。

-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 是空的

最新问题
© www.soinside.com 2019 - 2024. All rights reserved.