Node.js Streams：你需要知道的一切
图像来源
Node.js流以难以使用而闻名，甚至更难理解。好吧，我有个好消息 - 不再是这样了。
多年来，开发人员在那里创建了许多软件包，其唯一目的是简化流程。但在本文中，我将重点介绍本机Node.js流API。
“Streams是Node最好，也是最容易被误解的想法。”
- Dominic Tarr
什么是溪流？
流是数据的集合 - 就像数组或字符串一样。不同之处在于流可能无法一次全部可用，并且它们不必适合内存。这使得流真正强大的大量数据，或者数据这是一个从外部来源有人来工作时，大块的时间。
但是，流不仅仅是处理大数据。它们还为我们提供了代码中可组合性的强大功能。就像我们可以通过管道其他较小的Linux命令来组成强大的linux命令一样，我们可以在Node中使用流完全相同。
与Linux命令的可组合性
const grep = ... //用于grep输出的流
const wc = ... //用于wc输入的流
const wc = ... //用于wc输入的流grep.pipe（WC）Node中的许多内置模块实现了流接口：
从我的Pluralsight课程 - Advanced Node.js中捕获的截图
上面的列表提供了一些本机Node.js对象的示例，这些对象也是可读写的流。其中一些对象是可读写的流，如TCP套接字，zlib和加密流。
请注意，对象也是密切相关的。虽然HTTP响应是客户端上的可读流，但它是服务器上的可写流。这是因为在HTTP情况下，我们基本上从一个对象（http.IncomingMessage）读取并写入另一个（http.ServerResponse）。
另外要注意的是如何stdio流（stdin，stdout，stderr）有反流类型，当涉及到的子进程。这允许一种非常简单的方法来管理来自主流程stdio流的这些流。
一个实际的例子
理论很棒，但往往不是100％令人信服。让我们看一个示例，演示在内存消耗方面流可以在代码中产生的差异。
让我们先创建一个大文件：
const fs = require（'fs'）;
const file = fs.createWriteStream（'./ big.file'）; for（let i = 0; i <= 1e6; i ++）{ file.write（'Lorem ipsum dolor sit amet，consectetur adipisicing elit，sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.Ut enim ad minim veniam，quis nostrud练习ullamco laboris nisi ut aliquip ex ea commodo consequat.Duis aute irure dolor in repreptderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur。Excepteur sint occaecat cupidatat non proident，sunt in culpa qui officia deserunt mollit anim id est laborum。\ n' ）;
} file.end（）;
const file = fs.createWriteStream（'./ big.file'）; for（let i = 0; i <= 1e6; i ++）{ file.write（'Lorem ipsum dolor sit amet，consectetur adipisicing elit，sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.Ut enim ad minim veniam，quis nostrud练习ullamco laboris nisi ut aliquip ex ea commodo consequat.Duis aute irure dolor in repreptderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur。Excepteur sint occaecat cupidatat non proident，sunt in culpa qui officia deserunt mollit anim id est laborum。\ n' ）;
} file.end（）;看看我用来创建那个大文件的东西。一个可写的流！
该fs模块可用于使用流接口读取和写入文件。在上面的例子中，我们big.file通过带有循环的可写流100万行来写入。
运行上面的脚本会生成大约约400 MB的文件。
这是一个简单的Node Web服务器，专门用于big.file：
const fs = require（'fs'）;
const server = require（'http'）。createServer（）; server.on（'request'，（req，res）=> { fs.readFile（'./ big.file'，（err，data）=> { if（err）throw err; res.end（data）; }）;
}）; server.listen（8000）;
const server = require（'http'）。createServer（）; server.on（'request'，（req，res）=> { fs.readFile（'./ big.file'，（err，data）=> { if（err）throw err; res.end（data）; }）;
}）; server.listen（8000）;当服务器收到请求时，它将使用异步方法为大文件提供服务fs.readFile。但是，嘿，这不像我们阻止事件循环或任何事情。每件事都很棒，对吗？对？
好吧，让我们看看当我们运行服务器，连接到它并监视内存时会发生什么。
当我运行服务器时，它开始时具有正常的内存量，8.7 MB：
然后我连接到服务器。注意消耗的内存发生了什么：
哇 - 内存消耗跃升至434.8 MB。
在将big.file它们写入响应对象之前，我们基本上将整个内容放在内存中。这是非常低效的。
HTTP响应对象（res在上面的代码中）也是可写流。这意味着如果我们有一个表示内容的可读流big.file，我们可以将这两个相互管道并实现大致相同的结果，而不会消耗~400 MB的内存。
Node的fs模块可以为使用该createReadStream方法的任何文件提供可读流。我们可以将它传递给响应对象：
const fs = require（'fs'）;
const server = require（'http'）。createServer（）; server.on（'request'，（req，res）=> { const src = fs.createReadStream（'./ big.file'）; src.pipe（res）;}）; server.listen（8000）;
const server = require（'http'）。createServer（）; server.on（'request'，（req，res）=> { const src = fs.createReadStream（'./ big.file'）; src.pipe（res）;}）; server.listen（8000）;现在当你连接到这个服务器时，会发生一件神奇的事情（看一下内存消耗）：
发生了什么？
当客户端请求该大文件时，我们一次流一个块，这意味着我们根本不在内存中缓冲它。内存使用量增长了大约25 MB，就是这样。
您可以将此示例推到极限。big.file使用500万行而不是仅仅100万行重新生成，这将使文件超过2 GB，并且实际上大于Node中的默认缓冲区限制。
如果您尝试使用该文件fs.readFile，默认情况下您根本不能（您可以更改限制）。但是fs.createReadStream，对于请求者来说，将2 GB的数据流传输到请求者是没有问题的，最重要的是，进程内存使用量大致相同。
准备好学习流了吗？
本文是关于Node.js的Pluralsight课程的一部分。我在那里报道了类似的视频格式内容。
流101
Node.js中有四种基本流类型：可读，可写，双工和变换流。
- 可读流是可以从中消费数据的源的抽象。一个例子是
fs.createReadStream方法。 - 可写流是可以写入数据的目标的抽象。一个例子是
fs.createWriteStream方法。 - 双工流是可读和可写的。一个例子是TCP套接字。
- 变换流基本上是双工流，可用于在写入和读取数据时修改或转换数据。一个例子是
zlib.createGzip使用gzip压缩数据的流。您可以将转换流视为一个函数，其中输入是可写流部分，输出是可读流部分。您可能还会听到称为“ 直通流 ”的转换流。
所有流都是EventEmitter。它们发出可用于读取和写入数据的事件。但是，我们可以使用该pipe方法以更简单的方式使用流数据。
管道方法
这是你需要记住的神奇线条：
readableSrc .pipe（writableDest） .pipe（writableDest）在这个简单的行中，我们管道输出可读流 - 数据源，作为可写流的输入 - 目的地。源必须是可读流，目标必须是可写的。当然，它们也可以是双工/变换流。事实上，如果我们正在进入双工流，我们可以像在Linux中一样链接管道调用：
readableSrc .pipe（transformStream1）.pipe（transformStream2）.pipe（finalWrtitableDest）.pipe（transformStream1）.pipe（transformStream2）.pipe（finalWrtitableDest）该pipe方法返回目标流，这使我们能够进行上述链接。对于流a（读取），b和c（双面），和d（可写的），我们可以：
a.pipe（b）中.pipe（c）中.pipe（d）＃这相当于：
a.pipe（b）
b.pipe（c）
c.pipe（d）
a.pipe（b）
b.pipe（c）
c.pipe（d）＃在Linux中，相当于：
$ a | b | c | d
$ a | b | c | d该pipe方法是消费流的最简单方法。通常建议使用该pipe方法或使用事件消耗流，但避免混合这两者。通常，当您使用该pipe方法时，您不需要使用事件，但如果您需要以更自定义的方式使用流，则事件将是可行的方法。
流事件
除了从可读流源读取并写入可写目的地之外，该pipe方法还会自动管理一些事情。例如，它处理错误，文件结束以及一个流比另一个流更慢或更快的情况。
但是，流也可以直接与事件一起使用。这是该pipe方法主要用于读写数据的简化事件等效代码：
#readed.pipe（可写）readable.on（'data'，（chunk）=> { writable.write（chunk）;
}）;writable.write（chunk）;
}）;readable.on（'end'，（）=> { writable.end（）;
}）;writable.end（）;
}）;以下是可与可读写流一起使用的重要事件和函数的列表：
从我的Pluralsight课程 - Advanced Node.js中捕获的截图
事件和函数以某种方式相关，因为它们通常一起使用。
可读流上最重要的事件是：
data每当流将一大块数据传递给使用者时发出的事件- 该
end事件，在没有更多数据要从流中消耗时发出。
可写流上最重要的事件是：
- 该
drain事件，是可写流可以接收更多数据的信号。 - 该
finish事件，在将所有数据刷新到基础系统时发出。
可以组合事件和功能，以实现流的自定义和优化使用。要使用可读流，我们可以使用pipe/ unpipemethods或read/ unshift/ resume方法。要使用可写流，我们可以将它作为pipe/ 的目标unpipe，或者只是使用write方法写入它，并end在完成后调用方法。
可读流的暂停和流动模式
可读流有两种主要模式影响我们使用它们的方式：
- 它们可以处于暂停模式
- 或者在流动模式下
这些模式有时被称为拉动和推动模式。
默认情况下，所有可读流都以暂停模式启动，但它们可以轻松切换为流动，并在需要时返回暂停状态。有时，切换会自动进行。
当可读流处于暂停模式时，我们可以使用该read()方法根据需要从流中读取，但是，对于流动模式中的可读流，数据不断流动，我们必须监听事件以使用它。
在流动模式下，如果没有消费者可以处理数据，实际上可能会丢失数据。这就是为什么当我们在流动模式下有可读流时，我们需要一个data事件处理程序。实际上，只需添加一个data事件处理程序就可以将暂停的流切换为流动模式，并删除data事件处理程序会将流切换回暂停模式。其中一些是为了向后兼容旧的Node流接口而完成的。
要在这两种流模式之间手动切换，可以使用resume()和pause()方法。
从我的Pluralsight课程 - Advanced Node.js中捕获的截图
使用该pipe方法消耗可读流时，我们不必担心这些模式会pipe自动管理它们。
实现流
当我们在Node.js中讨论流时，有两个主要的不同任务：
- 实现流的任务。
- 消费它们的任务。
到目前为止，我们一直在谈论只消耗流。让我们实施一些！
流实现者通常require是stream模块的人。
实现可写流
要实现可写流，我们需要使用Writable流模块中的构造函数。
const {Writable} = require（'stream'）;我们可以通过多种方式实现可写流。例如，Writable如果需要，我们可以扩展构造函数
class myWritableStream扩展Writable { }}
但是，我更喜欢更简单的构造方法。我们只是从Writable构造函数创建一个对象，并传递一些选项。唯一需要的选项是write暴露要写入的数据块的函数。
const {Writable} = require（'stream'）;const outStream = new Writable（{ write（chunk，encoding，callback）{ console.log（chunk.toString（））; callback（）; } }）; process.stdin.pipe（outStream）;write（chunk，encoding，callback）{ console.log（chunk.toString（））; callback（）; } }）; process.stdin.pipe（outStream）;
这个write方法有三个参数。
- 该块通常是一个缓冲区，除非我们配置不同的数据流。
- 该编码参数，需要在这种情况下，但通常我们可以忽略它。
- 该回调是我们需要我们完成处理数据块之后调用一个函数。这就是写入是否成功的信号。要发出故障信号，请使用错误对象调用回调。
在outStream，我们只是console.log将块作为字符串，并callback在没有错误的情况下调用之后表示成功。这是一个非常简单且可能不那么有用的回声流。它将回应它收到的任何东西。
要使用这个流，我们可以简单地使用它process.stdin，这是一个可读的流，所以我们可以直接process.stdin进入我们的流outStream。
当我们运行上面的代码时，我们输入的任何内容都process.stdin将使用该outStream console.log行回显。
这不是一个非常有用的实现流，因为它实际上已经实现和内置。这非常相当于process.stdout。我们可以直接stdin进入stdout，我们将通过这一行获得完全相同的回声功能：
process.stdin.pipe（process.stdout）;实现可读流
要实现可读流，我们需要Readable接口，并从中构造一个对象，并read()在流的配置参数中实现一个方法：
const {Readable} = require（'stream'）;const inStream = new Readable（{ read（）{} }）;read（）{} }）;
有一种实现可读流的简单方法。我们可以直接push使用我们希望消费者使用的数据。
const {Readable} = require（'stream'）; const inStream = new Readable（{ read（）{} }）;read（）{} }）;inStream中。推（'ABCDEFGHIJKLM'）; inStream中。推（'NOPQRSTUVWXYZ'）;推（'ABCDEFGHIJKLM'）; inStream中。推（'NOPQRSTUVWXYZ'）;inStream中。推（null）; //没有更多数据推（null）; //没有更多数据
inStream.pipe（process.stdout）;当我们push成为一个null对象时，这意味着我们想要发信号通知该流没有更多数据。
要使用这个简单的可读流，我们可以简单地将其传输到可写流中process.stdout。
当我们运行上面的代码时，我们将从中读取所有数据inStream并将其回显到标准输出。很简单，但也不是很有效率。
我们基本上推流中的所有数据之前，它管道到process.stdout。当消费者要求时，更好的方法是按需推送数据。我们可以通过read()在配置对象中实现该方法来实现：
const inStream = new Readable（{ read（size）{ //对数据有需求......有人想读它。} }）;read（size）{ //对数据有需求......有人想读它。} }）;
当在可读流上调用read方法时，实现可以将部分数据推送到队列。例如，我们可以一次推送一个字母，从字符代码65（代表A）开始，并在每次推送时递增：
const inStream = new Readable（{ read（size）{ this.push（String.fromCharCode（this.currentCharCode ++））; if（this.currentCharCode> 90）{ this.push（null）; } } }）;read（size）{ this.push（String.fromCharCode（this.currentCharCode ++））; if（this.currentCharCode> 90）{ this.push（null）; } } }）;
inStream.currentCharCode = 65;inStream.pipe（process.stdout）;当消费者正在阅读可读流时，该read方法将继续触发，并且我们将推送更多字母。我们需要在某个地方停止这个循环，这就是当currentCharCode大于90（表示Z）时if语句推送null的原因。
这段代码相当于我们开始使用的更简单的代码，但现在我们在消费者要求时按需推送数据。你应该总是那样做。
实现双工/转换流
使用Duplex流，我们可以使用相同的对象实现可读和可写流。就像我们从两个接口继承一样。
这是一个示例双工流，它结合了上面实现的两个可写和可读示例：
const {Duplex} = require（'stream'）; const inoutStream = new Duplex（{ write（chunk，encoding，callback）{ console.log（chunk.toString（））; callback（）; }，read（size）{ this.push（String.fromCharCode（this.currentCharCode ++） ）; if（this.currentCharCode> 90）{ this.push（null）; } } }）; inoutStream.currentCharCode = 65;const inoutStream = new Duplex（{ write（chunk，encoding，callback）{ console.log（chunk.toString（））; callback（）; }，read（size）{ this.push（String.fromCharCode（this.currentCharCode ++） ）; if（this.currentCharCode> 90）{ this.push（null）; } } }）; inoutStream.currentCharCode = 65;
<strong>process.stdin.pipe（inoutStream）.pipe（process.stdout）;</strong>通过组合这些方法，我们可以使用此双工流来读取A到Z中的字母，我们也可以将其用作其回声功能。我们将可读stdin流传输到此双工流中以使用echo功能，我们将双工流本身stdout传输到可写流中以查看字母A到Z.
重要的是要理解双工流的可读和可写侧完全独立地操作。这仅仅是将两个特征分组到一个对象中。
变换流是更有趣的双工流，因为其输出是根据其输入计算的。
对于转换流，我们不必实现read或write方法，我们只需要实现一个transform结合它们的方法。它具有write方法的签名，我们也可以将它用于push数据。
这是一个简单的变换流，在将其转换为大写格式之后回显您输入的任何内容：
const {Transform} = require（'stream'）; const upperCaseTr = new Transform（{ transform（chunk，encoding，callback）{ this.push（chunk.toString（）。toUpperCase（））; callback（）; } }）; process.stdin.pipe（upperCaseTr）.pipe（process.stdout）;const upperCaseTr = new Transform（{ transform（chunk，encoding，callback）{ this.push（chunk.toString（）。toUpperCase（））; callback（）; } }）; process.stdin.pipe（upperCaseTr）.pipe（process.stdout）;
在这个我们正在消耗的变换流中，就像前面的双工流示例一样，我们只实现了一个transform()方法。在该方法中，我们将chunk其转换为大写版本，然后push将该版本转换为可读部分。
流对象模式
默认情况下，流期望缓冲区/字符串值。objectMode我们可以设置一个标志，让流接受任何JavaScript对象。
这是一个简单的例子来证明这一点。以下转换流组合使得一个功能可以将逗号分隔值的字符串映射到JavaScript对象中。因此“a,b,c,d”变得{a: b, c: d}。
const {Transform} = require（'stream'）;const commaSplitter = new Transform（{ readableObjectMode：true，readableObjectMode：true，transform（chunk，encoding，callback）{ this.push（chunk.toString（）。trim（）。split（'，'））; 打回来（）; } };this.push（chunk.toString（）。trim（）。split（'，'））; 打回来（）; } };const arrayToObject = new Transform（{ readableObjectMode：true，writableObjectMode：true，readableObjectMode：true，writableObjectMode：true，transform（chunk，encoding，callback）{ const obj = {}; for（let i = 0; i <chunk.length; i + = 2）{ obj [chunk [i]] = chunk [i + 1]; } this.push（obj）; 打回来（）; } };const obj = {}; for（let i = 0; i <chunk.length; i + = 2）{ obj [chunk [i]] = chunk [i + 1]; } this.push（obj）; 打回来（）; } };const objectToString = new Transform（{ writableObjectMode：true，writableObjectMode：true，transform（chunk，encoding，callback）{ this.push（JSON.stringify（chunk）+'\ n'）; 打回来（）; } };this.push（JSON.stringify（chunk）+'\ n'）; 打回来（）; } };process.stdin .pipe（commaSplitter）.pipe（arrayToObject）.pipe（objectToString ）.pipe（process.stdout）.pipe（commaSplitter）.pipe（arrayToObject）.pipe（objectToString ）.pipe（process.stdout）
我们传递输入字符串（例如“a,b,c,d”），通过commaSplitter该字符串将数组推送为可读数据（[“a”, “b”, “c”, “d”]）。readableObjectMode在该流上添加标志是必要的，因为我们在那里推送一个对象，而不是字符串。
然后我们将数组并将其arrayToObject传输到流中。我们需要一个writableObjectMode标志来使该流接受一个对象。它还会推送一个对象（映射到对象的输入数组），这就是为什么我们也需要readableObjectMode那里的标志。最后一个objectToString流接受一个对象，但推出一个字符串，这就是为什么我们只需要一个writableObjectMode标志。可读部分是普通字符串（字符串化对象）。
使用上面的例子
Node的内置转换流
Node有一些非常有用的内置变换流。即，zlib和加密流。
这是一个使用zlib.createGzip()流与fs可读/可写流相结合来创建文件压缩脚本的示例：
const fs = require（'fs'）;
const zlib = require（'zlib'）;
const file = process.argv [2]; fs.createReadStream（file）.pipe（zlib.createGzip（））。pipe（fs.createWriteStream（file +'.gz'））;
const zlib = require（'zlib'）;
const file = process.argv [2]; fs.createReadStream（file）.pipe（zlib.createGzip（））。pipe（fs.createWriteStream（file +'.gz'））;您可以使用此脚本将您传递的任何文件作为参数进行gzip。我们将该文件的可读流传输到zlib内置转换流中，然后传输到新gzip压缩文件的可写流中。简单。
使用管道的一个很酷的事情是，如果需要，我们实际上可以将它们与事件结合起来。比如说，我希望用户在脚本工作时看到进度指示器，在脚本完成时看到“完成”消息。由于该pipe方法返回目标流，我们也可以链接事件处理程序的注册：
const fs = require（'fs'）;
const zlib = require（'zlib'）;
const file = process.argv [2]; fs.createReadStream（file）.pipe（zlib.createGzip（））。
on（'data'，（）=> process.stdout.write（'。'））。pipe（fs.createWriteStream（file +'.zz' ））。on（'finish'，（）=> console.log（'完成'））;
const zlib = require（'zlib'）;
const file = process.argv [2]; fs.createReadStream（file）.pipe（zlib.createGzip（））。
on（'data'，（）=> process.stdout.write（'。'））。pipe（fs.createWriteStream（file +'.zz' ））。on（'finish'，（）=> console.log（'完成'））;因此，通过该pipe方法，我们可以轻松地使用流，但我们仍然可以使用需要的事件进一步自定义与这些流的交互。
这个pipe方法的优点在于我们可以用一种可读的方式逐个编写程序。例如，data我们可以简单地创建一个转换流来报告进度，而不是监听上面的事件，并.on()用另一个 .pipe()调用替换该 调用：
const fs = require（'fs'）;
const zlib = require（'zlib'）;
const file = process.argv [2]; const {Transform} = require（'stream'）; const reportProgress = new Transform（{ transform（chunk，encoding，callback）{ process.stdout.write（'。'）; callback（null，chunk）; }
}）; fs.createReadStream（file）.pipe（zlib.createGzip（））。
pipe（reportProgress）.pipe（fs.createWriteStream（file +'。。'））。on（'finish'，（）=> console.log（ '完成'））;
const zlib = require（'zlib'）;
const file = process.argv [2]; const {Transform} = require（'stream'）; const reportProgress = new Transform（{ transform（chunk，encoding，callback）{ process.stdout.write（'。'）; callback（null，chunk）; }
}）; fs.createReadStream（file）.pipe（zlib.createGzip（））。
pipe（reportProgress）.pipe（fs.createWriteStream（file +'。。'））。on（'finish'，（）=> console.log（ '完成'））;此reportProgress流是一个简单的直通流，但它也会将进度报告给标准输出。请注意我如何使用函数中的第二个参数callback()来推送transform()方法中的数据。这相当于首先推送数据。
组合流的应用是无止境的。例如，如果我们需要在gzip之前或之后加密文件，我们需要做的就是按照我们需要的确切顺序管道另一个转换流。我们可以使用Node的crypto模块：
const crypto = require（'crypto'）;
// ...// ...fs.createReadStream（file）.pipe（zlib.createGzip（））。
pipe（crypto.createCipher（'aes192'，'a_secret'））。
pipe（reportProgress）.pipe（fs.createWriteStream（file +'.zz'） ）.on（'finish'，（）=> console.log（'完成'））;.pipe（zlib.createGzip（））。
pipe（crypto.createCipher（'aes192'，'a_secret'））。
pipe（reportProgress）.pipe（fs.createWriteStream（file +'.zz'） ）.on（'finish'，（）=> console.log（'完成'））;上面的脚本压缩然后加密传递的文件，只有拥有秘密的人才能使用输出的文件。我们无法使用普通的解压缩实用程序解压缩此文件，因为它已加密。
要实际能够解压缩上面脚本压缩的任何内容，我们需要以相反的顺序使用相反的流加密和zlib，这很简单：
fs.createReadStream（文件）.pipe（加密createDecipher（ 'AES192'， 'a_secret'））.pipe（ZLIB。createGunzip（））.pipe（reportProgress）.pipe（fs.createWriteStream（file.slice（0， - 3）））。on（'finish'，（）=> console.log（'Done'））;.pipe（加密createDecipher（ 'AES192'， 'a_secret'））.pipe（ZLIB。createGunzip（））.pipe（reportProgress）.pipe（fs.createWriteStream（file.slice（0， - 3）））。on（'finish'，（）=> console.log（'Done'））;假设传递的文件是压缩版本，上面的代码将从中创建一个读取流，将其传递到加密createDecipher()流（使用相同的秘密），将其输出createGunzip()传递到zlib 流，然后将内容写回没有扩展部分的文件。
这就是我对这个话题的全部看法。谢谢阅读！直到下一次！
学习反应还是节点？查看我的书：
- 通过构建游戏了解React.js
- Node.js超越基础
![[转帖]USB-C和Thunderbolt 3连接线你搞懂了吗？---没搞明白.](https://resource.feng.com/resource/h062/h37/img2018112500332415.png)
