node.js - 连接两个（或 n 个）流

Question

• 2个流：

  给定可读流 stream1和，获取包含和连接stream2的流的惯用（简洁）方法是什么？stream1stream2

  我做不到stream1.pipe(outStream); stream2.pipe(outStream)，因为这样流的内容就混在一起了。

• n 个流：

  给定一个发出不确定数量的流的EventEmitter，例如

  eventEmitter.emit('stream', stream1)
  eventEmitter.emit('stream', stream2)
  eventEmitter.emit('stream', stream3)
  ...
  eventEmitter.emit('end')
  

  获取所有流连接在一起的流的惯用（简洁）方法是什么？

Answer 1

这可以用 vanilla nodejs 来完成

import { PassThrough } from 'stream'
const merge = (...streams) => {
    let pass = new PassThrough()
    let waiting = streams.length
    for (let stream of streams) {
        pass = stream.pipe(pass, {end: false})
        stream.once('end', () => --waiting === 0 && pass.emit('end'))
    }
    return pass
}

Answer 2

组合流包连接流。自述文件中的示例：

var CombinedStream = require('combined-stream');
var fs = require('fs');

var combinedStream = CombinedStream.create();
combinedStream.append(fs.createReadStream('file1.txt'));
combinedStream.append(fs.createReadStream('file2.txt'));

combinedStream.pipe(fs.createWriteStream('combined.txt'));

我相信您必须一次附加所有流。如果队列为空，则combinedStream自动结束。见问题#5。

stream-stream库是一个具有显式 的替代方案，.end但它不太受欢迎，并且可能没有经过充分测试。它使用 Node 0.10 的 streams2 API（参见这个讨论）。

Answer 3

如果您不关心流中数据的顺序，那么在nodejs中进行简单的reduce操作应该没问题！

const {PassThrough} = require('stream')

let joined = [s0, s1, s2, ...sN].reduce((pt, s, i, a) => {
  s.pipe(pt, {end: false})
  s.once('end', () => a.every(s => s.ended) && pt.emit('end'))
  return pt
}, new PassThrough())

干杯；）

Answer 4

现在可以使用异步迭代器轻松完成

async function* concatStreams(readables) {
  for (const readable of readables) {
    for await (const chunk of readable) { yield chunk }
  }
} 

你可以像这样使用它

const fs = require('fs')
const stream = require('stream')

const files = ['file1.txt', 'file2.txt', 'file3.txt'] 
const iterable = await concatStreams(files.map(f => fs.createReadStream(f)))

// convert the async iterable to a readable stream
const mergedStream = stream.Readable.from(iterable)

有关异步迭代器的更多信息：https ://2ality.com/2019/11/nodejs-streams-async-iteration.html

Answer 5

在 vanilla nodejs 中使用 ECMA 15+ 并结合Ivo和Feng的好答案。

该类PassThrough是一个微不足道的Transform流，它不会以任何方式修改流。

const { PassThrough } = require('stream');

const concatStreams = (streamArray, streamCounter = streamArray.length) => streamArray
  .reduce((mergedStream, stream) => {
    // pipe each stream of the array into the merged stream
    // prevent the automated 'end' event from firing
    mergedStream = stream.pipe(mergedStream, { end: false });
    // rewrite the 'end' event handler
    // Every time one of the stream ends, the counter is decremented.
    // Once the counter reaches 0, the mergedstream can emit its 'end' event.
    stream.once('end', () => --streamCounter === 0 && mergedStream.emit('end'));
    return mergedStream;
  }, new PassThrough());

可以这样使用：

const mergedStreams = concatStreams([stream1, stream2, stream3]);

Answer 6

您可能可以使其更简洁，但这是一个有效的方法：

var util = require('util');
var EventEmitter = require('events').EventEmitter;

function ConcatStream(streamStream) {
  EventEmitter.call(this);
  var isStreaming = false,
    streamsEnded = false,
    that = this;

  var streams = [];
  streamStream.on('stream', function(stream){
    stream.pause();
    streams.push(stream);
    ensureState();
  });

  streamStream.on('end', function() {
    streamsEnded = true;
    ensureState();
  });

  var ensureState = function() {
    if(isStreaming) return;
    if(streams.length == 0) {
      if(streamsEnded)
        that.emit('end');
      return;
    }
    isStreaming = true;
    streams[0].on('data', onData);
    streams[0].on('end', onEnd);
    streams[0].resume();
  };

  var onData = function(data) {
    that.emit('data', data);
  };

  var onEnd = function() {
    isStreaming = false;
    streams[0].removeAllListeners('data');
    streams[0].removeAllListeners('end');
    streams.shift();
    ensureState();
  };
}

util.inherits(ConcatStream, EventEmitter);

streams我们使用（流的队列；push到后面和shift从前面）、isStreaming和来跟踪状态streamsEnded。当我们得到一个新的流时，我们推送它，当一个流结束时，我们停止监听并转移它。当流结束时，我们设置streamsEnded.

在每个事件中，我们检查我们所处的状态。如果我们已经在流式传输（管道流），我们什么也不做。如果队列为空且streamsEnded已设置，则我们发出end事件。如果队列中有东西，我们恢复它并监听它的事件。

*请注意pause和resume是建议性的，因此某些流可能无法正确运行，并且需要缓冲。这个练习留给读者。

完成所有这些后，我将n=2通过构造一个EventEmitter，用它创建一个ConcatStream，然后发出两个stream事件，然后是一个end事件来完成这个案例。我敢肯定它可以做得更简洁，但我们不妨使用我们所拥有的。

Answer 7

https://github.com/joepie91/node-combined-stream2是组合流模块（如上所述）的插入式 Streams2 兼容替代品。它自动包装 Streams1 流。

组合流 2 的示例代码：

var CombinedStream = require('combined-stream2');
var fs = require('fs');

var combinedStream = CombinedStream.create();
combinedStream.append(fs.createReadStream('file1.txt'));
combinedStream.append(fs.createReadStream('file2.txt'));

combinedStream.pipe(fs.createWriteStream('combined.txt'));

Answer 8

这里两个最受好评的答案都不适用于异步流，因为它们只是通过管道传输东西，而不管源流是否准备好生成。我必须将内存中的字符串流与来自数据库的数据馈送结合起来，并且数据库内容始终位于结果流的末尾，因为它需要一秒钟才能获得数据库响应。这就是我最终为我的目的而写的内容。

export function joinedStream(...streams: Readable[]): Readable {
  function pipeNext(): void {
    const nextStream = streams.shift();
    if (nextStream) {
      nextStream.pipe(out, { end: false });
      nextStream.on('end', function() {
        pipeNext();
      });
    } else {
      out.end();
    }
  }
  const out = new PassThrough();
  pipeNext();
  return out;
}

Answer 9

streamee.js是一组基于 node 1.0+ 流的流转换器和作曲家，包括一个连接方法：

var stream1ThenStream2 = streamee.concatenate([stream1, stream2]);

Answer 10

下面的代码对我有用:)。已从之前给出的所有答案中获取输入

  const pipeStreams = (streams) => {
  const out = new PassThrough()
  // Piping the first stream to the out stream
  // Also prevent the automated 'end' event of out stream from firing
  streams[0].pipe(out, { end: false })
  for (let i = 0; i < streams.length - 2; i++) {
    // On the end of each stream (until the second last) pipe the next stream to the out stream
    // Prevent the automated 'end' event of out stream from firing
    streams[i].on('end', () => {
      streams[i + 1].pipe(out, { end: false })
    })
  }
  // On the end of second last stream pipe the last stream to the out stream.
  // Don't prevent the 'end flag from firing'
  streams[streams.length - 2].on('end', () => {
    streams[streams.length - 1].pipe(out)
  })
  return out
}