当前位置：Gxlcms > JavaScript > NodeStream的运行机制讲解（附示例）

NodeStream的运行机制讲解（附示例）

时间：2021-07-01 10:21:17 帮助过：17人阅读

本篇文章给大家带来的内容是关于Node Stream的运行机制讲解（附示例），有一定的参考价值，有需要的朋友可以参考一下，希望对你有所帮助。

如果你正在学习Node，那么流一定是一个你需要掌握的概念。如果你想成为一个Node高手，那么流一定是武功秘籍中不可缺少的一个部分。

引用自Stream-Handbook。由此可见，流对于深入学习Node的重要性。

流是什么？

你可以把流理解成一种传输的能力。通过流，可以以平缓的方式，无副作用的将数据传输到目的地。在Node中，Node Stream创建的流都是专用于String和Buffer上的，一般情况下使用Buffer。Stream表示的是一种传输能力，Buffer是传输内容的载体 (可以这样理解，Stream：外卖小哥哥， Buffer：你的外卖)。创建流的时候将ObjectMode设置true ，Stream同样可以传输任意类型的JS对象（除了null，null在流中有特殊用途）。

为什么要使用流？

现在有个需求，我们要向客户端传输一个大文件。如果采用下面的方式

  1. const fs = require('fs');
  2. const server = require('http').createServer();
  3. server.on('request', (req, res) => {
  4.   fs.readFile('./big.file', (err, data) => {
  5.     if (err) throw err;
  6.     res.end(data);
  7.   });
  8. });
  9. server.listen(8000);

每次接收一个请求，就要把这个大文件读入内存，然后再传输给客户端。通过这种方式可能会产生以下三种后果：

  • 内存耗尽

  • 拖慢其他进程

  • 增加垃圾回收器的负载

所以这种方式在传输大文件的情况下，不是一个好的方案。并发量一大，几百个请求过来很容易就将内存耗尽。

如果采用流呢？

  1. const fs = require('fs');
  2. const server = require('http').createServer();
  3. server.on('request', (req, res) => {
  4.   const src = fs.createReadStream('./big.file');
  5.   src.pipe(res);
  6. });
  7. server.listen(8000);

采用这种方式，不会占用太多内存，读取一点就传输一点，整个过程平缓进行，非常优雅。如果想在传输的过程中，想对文件进行处理，比如压缩、加密等等，也很好扩展（后面会具体介绍）。

流在Node中无处不在。从下图中可以看出：

20181008153896256368933.png

Stream分类

Stream分为四大类：

  • Readable（可读流）

  • Writable （可写流）

  • Duplex （双工流）

  • Transform （转换流）

Readable

可读流中的数据，在以下两种模式下都能产生数据。

  • Flowing Mode

  • Non-Flowing Mode

两种模式下，触发的方式以及消耗的方式不一样。

Flowing Mode：数据会源源不断地生产出来，形成“流动”现象。监听流的data事件便可进入该模式。

Non-Flowing Mode下：需要显示地调用read()方法，才能获取数据。

两种模式可以互相转换

3723356508-5bcea6087de1b_articlex.png

流的初始状态是Null，通过监听data事件，或者pipe方法，调用resume方法，将流转为Flowing Mode状态。Flowing Mode状态下调用pause方法，将流置为Non-Flowing Mode状态。Non-Flowing Mode状态下调用resume方法，同样可以将流置为Flowing Mode状态。

下面详细介绍下两种模式下，Readable流的运行机制。

Flowing Mode

在Flowing Mode状态下，创建的myReadable读流，直接监听data事件，数据就源源不断的流出来进行消费了。

  1. myReadable.on('data',function(chunk){
  2.       consume(chunk);//消费流
  3. })

一旦监听data事件之后，Readable内部的流程如下图所示

3074819627-5bcea6073397a_articlex.png

核心的方法是流内部的read方法，它在参数n为不同值时，分别触发不同的操作。下面描述中的hightwatermark表示的是流内部的缓冲池的大小。

  • n=undefined（消费数据，并触发一次可读流）

  • n=0（触发一次可读流，但是不会消费）

  • n>hightwatermark（修改hightwatermark的值）

  • n<buffer的总数据数（直接返回n个字节的数据）

  • n>buffer (可以返回null，也可以返回buffer所有的数据（当时最后一次读取）)

图中黄色标识的_read()，是用户实现流所需要自己实现的方法，这个方法就是实际读取流的方式（可以这样理解，外卖平台给你提供外卖的能力，那_read()方法就相当于你下单点外卖）。后面会详细介绍如何实现_read方法。

以上的流程可以描述为：监听data方法，Readable内部就会调用read方法，来进行触发读流操作，通过判断是同步还是异步读取，来决定读取的数据是否放入缓冲区。如果为异步的，那么就要调用flow方法，来继续触发read方法，来读取流，同时根据size参数判定是否emit('data')来消费流，循环读取。如果是同步的，那就emit('data')来消费流，同时继续触发read方法，来读取流。一旦push方法传入的是null，整个流就结束了。

从使用者的角度来看，在这种模式下，你可以通过下面的方式来使用流

  1. const fs = require('./fs');
  2. const readFile = fs.createReadStream('./big.file');
  3. const writeFile = fs.createWriteStream('./writeFile.js');
  4. readFile.on('data',function(chunk){
  5.       writeFile1.write(chunk);
  6. })
Non-Flowing Mode

相对于Flowing mode，Non-Flowing Mode要相对简单很多。

消费该模式下的流，需要使用下面的方式

  1. myReadable.on(‘readable’,function(){
  2.      const chunk = myReadable.read()
  3.      consume(chunk);//消费流
  4. })

在Non-Flowing Mode下，Readable内部的流程如下图：

3104384196-5bcea6072c53f_articlex.png

从这个图上看出，你要实现该模式的读流，同样要实现一个_read方法。

整个流程如下：监听readable方法，Readable内部就会调用read方法。调用用户实现的_read方法，来push数据到缓冲池，然后发送emit readable事件，通知用户端消费。

从使用者的角度来看，你可以通过下面的方式来使用该模式下的流

  1. const fs = require('fs');
  2. const readFile = fs.createReadStream('./big.file');
  3. const writeFile = fs.createWriteStream('./writeFile.js');
  4. readFile.on('readable',function(chunk) {
  5.     while (null !== (chunk = myReadable.read())) {
  6.         writeFile.write(chunk);
  7.     }
  8. });

Writable

相对于读流，写流的机制就更容易理解了。

写流使用下面的方式进行数据写入

  1. myWrite.write(chunk);

调用write后，内部Writable的流程如下图所示

3913744986-5bcea6072c5c8_articlex.png

类似于读流，实现一个写流，同样需要用户实现一个_write方法。

整个流程是这样的：调用write之后，会首先判定是否要写入缓冲区。如果不需要，那就调用用户实现的_write方法，将流写入到相应的地方，_write会调用一个writeable内部的一个回调函数。

从使用者的角度来看，使用一个写流，采用下面的代码所示的方式。

  1. const fs = require('fs');
  2. const readFile = fs.createReadStream('./big.file');
  3. const writeFile = fs.createWriteStream('./writeFile.js');
  4. readFile.on('data',function(chunk) {
  5.     writeFile.write(chunk);
  6. })

可以看到，使用写流是非常简单的。

我们先讲解一下如何实现一个读流和写流，再来看Duplex和Transform是什么，因为了解了如何实现一个读流和写流，再来理解Duplex和Transform就非常简单了。

实现自定义的Readable

实现自定义的Readable，只需要实现一个_read方法即可，需要在_read方法中调用push方法来实现数据的生产。如下面的代码所示：

  1. const Readable = require('stream').Readable;
  2. class MyReadable extends Readable {
  3.     constructor(dataSource, options) {
  4.         super(options);
  5.         this.dataSource = dataSource;
  6.     }
  7.     _read() {
  8.         const data = this.dataSource.makeData();
  9.         setTimeout(()=>{
  10.             this.push(data);
  11.         });
  12.     }
  13. }
  14. // 模拟资源池
  15. const dataSource = {
  16.     data: new Array(10).fill('-'),
  17.     makeData() {
  18.         if (!dataSource.data.length) return null;
  19.         return dataSource.data.pop();
  20.     }
  21. };
  22. const myReadable = new MyReadable(dataSource,);
  23. myReadable.on('readable', () => {
  24.     let chunk;
  25.     while (null !== (chunk = myReadable.read())) {
  26.         console.log(chunk);
  27.     }
  28. });

实现自定义的writable

实现自定义的writable，只需要实现一个_write方法即可。在_write中消费chunk写入到相应地方，并且调用callback回调。如下面代码所示：

  1. const Writable = require('stream').Writable;
  2. class Mywritable extends  Writable{
  3.     constuctor(options){
  4.         super(options);
  5.     }
  6.     _write(chunk,endcoding,callback){
  7.         console.log(chunk);
  8.         callback && callback();
  9.     }
  10. }
  11. const myWritable = new Mywritable();

Duplex

双工流：简单理解，就是讲一个Readable流和一个Writable流绑定到一起，它既可以用来做读流，又可以用来做写流。

实现一个Duplex流，你需要同时实现_read和_write方法。

有一点需要注意的是：它所包含的 Readable流和Writable流是完全独立，互不影响的两个流，两个流使用的不是同一个缓冲区。通过下面的代码可以验证

  1. // 模拟资源池1
  2. const dataSource1 = {
  3.     data: new Array(10).fill('a'),
  4.     makeData() {
  5.         if (!dataSource1.data.length) return null;
  6.         return dataSource1.data.pop();
  7.     }
  8. };
  9. // 模拟资源池2
  10. const dataSource2 = {
  11.     data: new Array(10).fill('b'),
  12.     makeData() {
  13.         if (!dataSource2.data.length) return null;
  14.         return dataSource2.data.pop();
  15.     }
  16. };
  17. const Readable = require('stream').Readable;
  18. class MyReadable extends Readable {
  19.     constructor(dataSource, options) {
  20.         super(options);
  21.         this.dataSource = dataSource;
  22.     }
  23.     _read() {
  24.         const data = this.dataSource.makeData();
  25.         setTimeout(()=>{
  26.             this.push(data);
  27.         })
  28.     }
  29. }
  30. const Writable = require('stream').Writable;
  31. class MyWritable extends Writable{
  32.     constructor(options){
  33.         super(options);
  34.     }
  35.     _write(chunk, encoding, callback) {
  36.         console.log(chunk.toString());
  37.         callback && callback();
  38.     }
  39. }
  40. const Duplex = require('stream').Duplex;
  41. class MyDuplex extends Duplex{
  42.     constructor(dataSource,options) {
  43.         super(options);
  44.         this.dataSource = dataSource;
  45.     }
  46.     _read() {
  47.         const data = this.dataSource.makeData();
  48.         setTimeout(()=>{
  49.             this.push(data);
  50.         })
  51.     }
  52.     _write(chunk, encoding, callback) {
  53.         console.log(chunk.toString());
  54.         callback && callback();
  55.     }
  56. }
  57. const myWritable = new MyWritable();
  58. const myReadable = new MyReadable(dataSource1);
  59. const myDuplex = new MyDuplex(dataSource1);
  60. myReadable.pipe(myDuplex).pipe(myWritable);

打印的结果是

  1. abababababababababab

从这个结果可以看出，myReadable.pipe(myDuplex)，myDuplex充当的是写流，写入的内容是a；myDuplex.pipe(myWritable)，myDuplex充当的是读流，往myWritable写的却是b；所以说它所包含的 Readable流和Writable流是完全独立的。

Transform

理解了Duplex，就更好理解Transform了。Transform是一个转换流，它既有读的功能又有写的功能，但是它和Duplex不同的是，它的读流和写流共用同一个缓冲区；也就是说，通过它读入什么，那它就能写入什么。

实现一个Transform，你只需要实现一个_transform方法。比如最简单的Transform:PassThrough，其源代码如下所示

2845945438-5bcea60761634_articlex.png

PassThrough就是一个Transform，但是这个转换流，什么也没做，相当于一个透明的转换流。可以看到_transform中什么都没有，只是简单的将数据进行回调。

如果我们在这个环节做些扩展，只需要在_transform中直接扩展就行了。比如我们可以对流进行压缩，加密，混淆等等操作。

BackPress

最后介绍一个流中非常重要的一个概念：背压。要了解这个，我们首先来看下pipe和highWaterMaker是什么。

pipe

首先看下下面的代码

  1. const fs = require('./fs');
  2. const readFile = fs.createReadStream('./big.file');
  3. const writeFile = fs.createWriteStream('./writeFile.js');
  4. readFile.pipe(writeFile);

上面的代码和下面是等价的

  1. const fs = require('./fs');
  2. const readFile = fs.createReadStream('./big.file');
  3. const writeFile = fs.createWriteStream('./writeFile.js');
  4. readFile.on('data',function(data){
  5.     var flag = ws.write(data);
  6.     if(!flag){ // 当前写流缓冲区已满，暂停读数据
  7.         readFile.pause();
  8.     }
  9. })
  10. writeFile.on('drain',function()){
  11.     readFile.resume();// 当前写流缓冲区已清空，重新开始读流
  12. }
  13. readFile.on('end',function(data){
  14.     writeFile.end();//将写流缓冲区的数据全部写入，并且关闭写入的文件
  15. })

pipe所做的操作就是相当于为写流和读流自动做了速度的匹配。

读写流速度不匹配的情况下，一般情况下不会造成什么问题，但是会造成内存增加。内存消耗增加，就有可能会带来一系列的问题。所以在使用的流的时候，强烈推荐使用pipe。

highWaterMaker

highWaterMaker说白了，就是定义缓冲区的大小。

  • 默认16Kb（Readable最大8M)

  • 可以自定义

背压的概念可以理解为：为了防止读写流速度不匹配而产生的一种调整机制；背压该调整机制的触发时机，受限于highWaterMaker设置的大小。

如上面的代码 var flag = ws.write(data);，一旦写流的缓冲区满了，那flag就会置为false，反向促进读流的速度调整。

Stream的应用场景

主要有以下场景

  1. 文件操作(复制，压缩，解压，加密等)

下面的就很容易就实现了文件复制的功能。

  1. const fs = require('fs');
  2. const readFile = fs.createReadStream('big.file');
  3. const writeFile = fs.createWriteStream('big_copy.file');
  4. readFile.pipe(writeFile);

那我们想在复制的过程中对文件进行压缩呢？

  1. const fs = require('fs');
  2. const readFile = fs.createReadStream('big.file');
  3. const writeFile = fs.createWriteStream('big.gz');
  4. const zlib = require('zlib');
  5. readFile.pipe(zlib.createGzip()).pipe(writeFile);

实现解压、加密也是类似的。

  1. 静态文件服务器

比如需要返回一个html，可以使用如下代码。

  1. var http = require('http');
  2. var fs = require('fs');
  3. http.createServer(function(req,res){
  4.     fs.createReadStream('./a.html').pipe(res);
  5. }).listen(8000);

以上就是Node Stream的运行机制讲解（附示例）的详细内容，更多请关注Gxl网其它相关文章！

人气教程排行

本站所有资源全部来源于网络，若本站发布的内容侵害到您的隐私或者利益，请联系我们删除！