Node.js：将可读流管道拆分为多个顺序可写流答案

【问题标题】：Node.js: splitting a readable stream pipe to multiple sequential writable streamsNode.js：将可读流管道拆分为多个顺序可写流
【发布时间】：2014-02-13 17:56:14
【问题描述】：

给定Readable 流（可能是process.stdin 或文件流），将pipe() 用于自定义Writable 流是否可能/实用，该流将填充子Writable 直到一定大小;然后关闭该子流；打开一个新的Writable 流并继续？

（上下文是将大量数据从管道上传到 CDN，然后将其分成合理大小的块，而无需先将数据写入磁盘。）

我尝试创建一个Writable 来处理_write 函数中子流的打开和关闭，但是当传入的块太大而无法容纳现有的子流时问题就来了：它有将一些块写入旧流；创建新流；然后在完成_write 调用之前等待新流上的open 事件。

我的另一个想法是创建一个额外的Duplex 或Transform 流来缓冲管道并确保进入Writable 的块绝对等于或小于现有子流可以的数量接受，给Writable 时间来改变子流。

或者，这是否会使所有事情变得过于复杂，并且有一种更简单的方法来完成原始任务？

【问题讨论】：

标签： node.js stream

【解决方案1】：

在寻找相关问题的答案时，我碰到了这个问题。如何解析文件并根据行中的某些类别值将其拆分为单独的文件。

我已尽力更改我的代码，使其与您的问题更相关。然而，这很快就适应了。未测试。将其视为伪代码。

var fs = require('fs'),
  through = require('through');

 var destCount = 0, dest, size = 0, MAX_SIZE = 1000;

readableStream
  .on('data', function(data) {
    var out = data.toString() + "\n";
    size += out.length;
    if(size > MAX_SIZE) {
      dest.emit("end");
      dest = null;
      size = 0;
    }
    if(!dest) {
      // option 1. manipulate data before saving them.
      dest = through();
      dest.pipe(fs.createWriteStream("log" + destCount))
      // option 2. write directly to file
      // dest = fs.createWriteStream("log" + destCount);
    }
    dest.emit("data", out);
  })
  .on('end', function() {
    dest.emit('end');
  });

【讨论】：

【解决方案2】：

我会在Readable 和Writable 流之间引入一个Transform。在它的_transform 中，我会做我需要的所有逻辑。

也许，我只有一个 Readable 和一个 Transform。 _transform 方法将创建我需要的所有可写流

就个人而言，我仅在将数据转储到某处并且处理完该块时才使用可写流。
我尽可能避免实现_read 和_write 并滥用Transform 流。

但是我在你的问题中不明白的一点是关于大小。什么意思？

【讨论】：

目标是不断地从Readable 流中读取最多1GB 的数据，并通过Writable 流。达到 1GB 后，需要关闭目标 Writable 流并打开另一个流，然后该过程继续，从其停止的相同 Readable 流中读取。由于数据高达5GB（CDN的限制），并且涉及的云服务器磁盘和内存空间有限，我不想缓冲它然后打开一个Writable：它需要是一个稳定的流量。