NodeJS Stream

忘是亡心i 2022-06-02 07:14 120阅读 0赞

## 什么是 Stream ##

在 Unix 系统中流就是一个很常见也很重要的概念，从术语上讲流是对输入输出设备的抽象。

ls | grep *.js

类似这样的代码我们在写脚本的时候经常可以遇到，使用  **`|` **连接两条命令，把前一个命令的结果作为后一个命令的参数传入，这样数据像是水流在管道中传递，每个命令类似一个处理器，对数据做一些加工，因此 **|** 被称为 “ **管道符号**”。

## NodeJS 中 Stream 的几种类型 ##

从程序角度而言流是 **有方向的数据**，按照流动方向可以分为三种流

1.  设备流向程序：readable
2.  程序流向设备：writable
3.  双向：duplex、transform

NodeJS 关于流的操作被封装到了  [Stream][] 模块，这个模块也被多个核心模块所引用。按照 Unix 的哲学：一切皆文件，在 NodeJS 中对文件的处理多数使用流来完成

1.  普通文件
2.  设备文件（stdin、stdout）
3.  网络文件（http、net）

下面来几个小例子

### **1、从另外一个文件读取，然后再写入当前的文件中**    ###

var fs = require('fs');
    
    //readFileSync 声名一个可读可写的流
    var source = fs.readFileSync('../Buffer/img.jpg'); // 读出文件流的数据 
    
    //再写进当前的文件中
    fs.writeFileSync('stream_copy_img.jpg' , source);

这种方法太爆力，直接读直接写，所以在读取文件流 和 写入文件流 的时候是有监听事件的

### 2、读取的监听事件 ###

执行后走过了 data、 readable、 end、 close 的监听，在data 里还打印了读取的数据

![SouthEast][]

### 3、而在传递过程中是可 **暂定** ，然后一会再接着开始 ###

在 data 里面监听到有数据传递的时候用 readStream.pause()   暂停，然后 3秒种后用 readStream.resume();  接着开始，这里是走了一次。

### 4、写入的监听    ###

data 事件里传递的数据如果是大文件并不是次就能传递完的，会分次传递，然后等到传递完，如果在 data 里直接加写会的话是读取完了才会写，大文件如果全部读出来再写，有N多次请求的时候可能会把内存都爆满了，所以最好是**边读边写 边读边写 边读边写**

下面是把一个文件 da\_img.jpg 读取到再写一个同样的文件 new\_da\_img.jpg 在当前文件中

当 data 里监听到有数据进来的时候用 **writeStream.write(chunk)** 开始写入  **=== false** 就是第二次有数据传递进来的时候我第一次还没写完，然后就用 **readStream.pause()**先暂停一会，等我写完先，然后再监听写入的事件 **writeStream** 的 ‘ **drain**’方法，写完后再继续开始读取

### **pipe**    ###

**其实上面的小例子，让文件  边读边写 系统自带有一个  ****pipe** **方法，可以完成以上的代码，两行代码就可以搞定。哈哈哈……**

var fs = require('fs');
    //读一个大文件 da_img.jpg ，然后把写放方法放到 pipe 中,pipe直接帮我们完成了边读边写的功能
     fs.createReadStream('da_img.jpg').pipe(fs.createWriteStream('1-pipe.jpg'));

在读取文件和写入文件的时候，系统的方法往往不能满足项目的一些需求

读写的原理是：

可读流是负责读取外部的数据，并把数据缓存到内部的 buffer 数组

可写流负责消费数据，从可读流里获取到数据，然后对得到的数据块进行处理，怎么处理就取决于内部的 \_write 方法，我们可以重写  \_write 方法可以看 下面的小例子

var Readable = require('stream').Readable //拿到 可读流的构造函数
    var Writable = require('stream').Writable //拿到 可写流的构造函数
    
    var readStream = new Readable()   //new 一个可读流实例
    var writStream = new Writable()   //new 一个可写流实例
    
    readStream.push('I ')    // 对可读流 readStream 添加一些数据
    readStream.push('Love ')
    readStream.push('Imooc\n')
    readStream.push(null)
    
    //重写 可写流的 _write方法
    writStream._write = function(chunk, encode, cb){
    	console.log(chunk.toString())
    	cb()
    }
    
    //最后用 pipe 把 可读流 和 可写流 连接起来
    readStream.pipe(writStream)

下面来个更详细的示例

示例的内容是自己定制 可读流  可写流   转换流的方法，然后通过 util 让自己定制的 可读流、可写流、转换流 **继承原型里的 可读流、可写流、转换流 方法并改掉重写一些里面的方法，最后用 pipe 把它们连接起来，  让它更适合自己用**

var stream = require('stream')  //拿到 stream
    var util = require('util')  //拿到 工具类
    
    // 自己定制的可读流
    	function ReadStream(){
    		// 改变它的上下文，让它可以调用 可读类的方法
    		stream.Readable.call(this)
    	}
    
    	// 通过 util.inherits 来声名 自己定制的可读流 继承可读流里面原型
    	util.inherits(ReadStream, stream.Readable)
    
    	//为可读流添加原型链上的 _read 方法
    	//可读流就是往里面添加数据 
    	ReadStream.prototype._read = function(){
    		readStream.push('I ')    // 对可读流 readStream 添加一些数据
    		readStream.push('Love ')
    		readStream.push('Imooc\n')
    		readStream.push(null
    	}
    
    
    // 自己定制的可写流
    	function WritStream(){
    		// 改变它的上下文，让它可以调用 可读类的方法
    		stream.Writable.call(this)
    		this._cached = new Buffer('')
    	}
    	// 通过 util.inherits 来声名 自己定制的可写流 继承可写流里面原型
    	util.inherits(WritStream, stream.Writable)
    
    	//为可写流添加原型链上的 _write 方法 (即重写掉 _write 方法)
    	WritStream.prototype._write = function(chunk, encode, cb){
    		console.log(chunk.toString())
    		cb()
    	}
    
    
    // 自己定制的转换流
    	function TransformStream(){
    		// 改变它的上下文，让它可以调用 转换流的方法
    		stream.Transform.call(this)
    	}
    	// 通过 util.inherits 来声名 自己定制的转换流 继承转换流里面原型
    	util.inherits(TransformStream, stream.Transform)
    
    	//为转换流添加原型链上的 transform 方法 (即重写掉 _transform 方法)
    	TransformStream.prototype._transform = function(chunk, encode, cb){
    		console.log(chunk.toString())
    		cb()
    	}
    
    	//重点是 _flush
    	TransformStream.prototype._flush = function(chunk, encode, cb){
    		//为读到的数据加一些额外的内容
    		this.push('Oh yes')
    		cb()
    	}
    
    //生成数据 
    
    //1、
    var rs = new ReadStream()  // new 可读流
    var ws = new WritStream()  // new 可写流
    var ts = new TransformStream()  // new 转化流
    
    //2、把读到的数据 pipe 给转换流，让它对内容进行额外的定制和处理，然后再pipe给可写流打印
    rs.pipe(ts).pipe(ws)

[Stream]: https://link.zhihu.com/?target=http%3A//nodejs.org/docs/latest/api/stream.html
[SouthEast]: /images/20220602/e409a959aec748728efb57bbe4f3edd2.png