浅谈Node.js：理解stream
Stream在node.js中是一个抽象的接口，基于EventEmitter，也是一种Buffer的高级封装，用来处理流数据。流模块便是提供各种API让我们可以很简单的使用Stream。
流分为四种类型，如下所示：
- Readable，可读流
- Writable，可写流
- Duplex，读写流
- Transform，扩展的Duplex，可修改写入的数据
1、Readable可读流
通过stream.Readable可创建一个可读流，它有两种模式：暂停和流动。
在流动模式下，将自动从下游系统读取数据并使用data事件输出；暂停模式下，必须显示调用stream.read()方法读取数据，并触发data事件。
所有的可读流最开始都是暂停模式，可以通过以下方法切换到流动模式：
- 监听'data'事件
- 调用stream.resume()方法
- 调用stream.pipe()方法将数据输出到一个可写流Writable
同样地，也可以切换到暂停模式，有两种方法：
- 如果没有设置pipe目标，调用stream.pause()方法即可。
- 如果设置了pipe目标，则需要移除所有的data监听和调用stream.unpipe()方法
在Readable对象中有一个_readableSate的对象，通过该对象可以得知流当前处于什么模式，如下所示：
- readable._readableState.flowing=null，没有数据消费者，流不产生数据
- readable._readableState.flowing=true，处于流动模式
- readable._readableState.flowing=false，处于暂停模式
为什么使用流取数据
对于小文件，使用fs.readFile()方法读取数据更方便，但需要读取大文件的时候，比如几G大小的文件，使用该方法将消耗大量的内存，甚至使程序崩溃。这种情况下，使用流来处理是更合适的，采用分段读取，便不会造成内存的'爆仓'问题。
data事件
在stream提供数据块给消费者时触发，有可能是切换到流动模式的时候，也有可能是调用readable.read()方法且有有效数据块的时候，使用如下所示：
constfs=require('fs');
constrs=fs.createReadStream('./appbak.js');
varchunkArr=[],
chunkLen=0;
rs.on('data',(chunk)=>{
chunkArr.push(chunk);
chunkLen+=chunk.length;
});
rs.on('end',(chunk)=>{
console.log(Buffer.concat(chunkArr,chunkLen).toString());
});
readable事件
当流中有可用数据能被读取时触发，分为两种，新的可用的数据和到达流的末尾，前者stream.read()方法返回可用数据，后者返回null，如下所示：
constrs=fs.createReadStream('./appbak.js');
varchunkArr=[],
chunkLen=0;
rs.on('readable',()=>{
varchunk=null;
//这里需要判断是否到了流的末尾
if((chunk=rs.read())!==null){
chunkArr.push(chunk);
chunkLen+=chunk.length;
}
});
rs.on('end',(chunk)=>{
console.log(Buffer.concat(chunkArr,chunkLen).toString());
});
pause和resume方法
stream.pause()方法让流进入暂停模式，并停止'data'事件触发，stream.resume()方法使流进入流动模式，并恢复'data'事件触发，也可以用来消费所有数据，如下所示：
constrs=fs.createReadStream('./下载.png');
rs.on('data',(chunk)=>{
console.log(`接收到${chunk.length}字节数据...`);
rs.pause();
console.log(`数据接收将暂停1.5秒.`);
setTimeout(()=>{
rs.resume();
},1000);
});
rs.on('end',(chunk)=>{
console.log(`数据接收完毕`);
});
pipe(destination[,options])方法
pipe()方法绑定一个可写流到可读流上，并自动切换到流动模式，将所有数据输出到可写流，以及做好了数据流的管理，不会发生数据丢失的问题，使用如下所示：
constrs=fs.createReadStream('./app.js');
rs.pipe(process.stdout);
以上介绍了多种可读流的数据消费的方法，但对于一个可读流，最好只选择其中的一种，推荐使用pipe()方法。
2、Writable可写流
所有的可写流都是基于stream.Writable类创建的，创建之后便可将数据写入该流中。
write(chunk[,encoding][,callback])方法
write()方法向可写流中写入数据，参数含义：
- chunk，字符串或buffer
- encoding，若chunk为字符串，则是chunk的编码
- callback，当前chunk数据写入磁盘时的回调函数
该方法的返回值为布尔值，如果为false，则表示需要写入的数据块被缓存并且此时缓存的大小超出highWaterMark阀值，否则为true。
使用如下所示：
constws=fs.createWriteStream('./test.txt');
ws.write('nihao','utf8',()=>{process.stdout.write('thischunkisflushed.');});
ws.end('done.')
背压机制
如果可写流的写入速度跟不上可读流的读取速度，write方法添加的数据将被缓存，逐渐增多，导致占用大量内存。我们希望的是消耗一个数据，再去读取一个数据，这样内存就维持在一个水平上。如何做到这一点？可以利用write方法的返回值来判断可写流的缓存状态和'drain'事件，及时切换可读流的模式，如下所示：
functioncopy(src,dest){
src=path.resolve(src);
dest=path.resolve(dest);
constrs=fs.createReadStream(src);
constws=fs.createWriteStream(dest);
console.log('正在复制中...');
conststime=+newDate();
rs.on('data',(chunk)=>{
if(null===ws.write(chunk)){
rs.pause();
}
});
ws.on('drain',()=>{
rs.resume();
});
rs.on('end',()=>{
constetime=+newDate();
console.log(`已完成，用时：${(etime-stime)/1000}秒`);
ws.end();
});
functioncalcProgress(){
}
}
copy('./CSS权威指南第3版.pdf','./javascript.pdf');
drain事件
如果Writable.write()方法返回false，则drain事件将会被触发，上面的背压机制已经使用了该事件。
finish事件
在调用stream.end()方法之后且所有缓存区的数据都被写入到下游系统，就会触发该事件，如下所示：
constws=fs.createWriteStream('./alphabet.txt');
constalphabetStr='abcdefghijklmnopqrstuvwxyz';
ws.on('finish',()=>{
console.log('done.');
});
for(letletterofalphabetStr.split()){
ws.write(letter);
}
ws.end();//必须调用
end([chunk][,encoding][,callback])方法
end()方法被调用之后，便不能再调用stream.write()方法写入数据，负责将抛出错误。
3、Duplex读写流
Duplex流同时实现了Readable与Writable类的接口，既是可读流，也是可写流。例如'zlibstreams'、'cryptostreams'、'TCPsockets'等都是Duplex流。
4、Transform流
Duplex流的扩展，区别在于，Transform流自动将写入端的数据变换后添加到可读端。例如：'zlibstreams'、'cryptostreams'等都是Transform流。
5、四种流的实现
stream模块提供的API可以让我们很简单的实现流，该模块使用require('stream')引用，我们只要继承四种流中的一个基类(stream.Writable,stream.Readable,stream.Duplex,orstream.Transform)，然后实现它的接口就可以了，需要实现的接口如下所示：
|Use-case|Class|Method(s)toimplement|
|-------------|-------------|-----|
|Readingonly|Readable|_read|
|Writingonly|Writable|_write,_writev|
|Readingandwriting|Duplex|_read,_write,_writev|
|Operateonwrittendata,thenreadtheresult|Transform|_transform,_flush|
Readable流实现
如上所示，我们只要继承Readable类并实现_read接口即可，，如下所示：
constReadable=require('stream').Readable;
constutil=require('util');
constalphabetArr='abcdefghijklmnopqrstuvwxyz'.split();
/*functionAbReadable(){
if(!thisinstanceofAbReadable){
returnnewAbReadable();
}
Readable.call(this);
}
util.inherits(AbReadable,Readable);
AbReadable.prototype._read=function(){
if(!alphabetArr.length){
this.push(null);
}else{
this.push(alphabetArr.shift());
}
};
constabReadable=newAbReadable();
abReadable.pipe(process.stdout);*/
/*classAbReadableextendsReadable{
constructor(){
super();
}
_read(){
if(!alphabetArr.length){
this.push(null);
}else{
this.push(alphabetArr.shift());
}
}
}
constabReadable=newAbReadable();
abReadable.pipe(process.stdout);*/
/*constabReadable=newReadable({
read(){
if(!alphabetArr.length){
this.push(null);
}else{
this.push(alphabetArr.shift());
}
}
});
abReadable.pipe(process.stdout);*/
constabReadable=Readable();
abReadable._read=function(){
if(!alphabetArr.length){
this.push(null);
}else{
this.push(alphabetArr.shift());
}
}
abReadable.pipe(process.stdout);
以上代码使用了四种方法创建一个Readable可读流，必须实现_read()方法，以及用到了readable.push()方法，该方法的作用是将指定的数据添加到读取队列。
Writable流实现
我们只要继承Writable类并实现_write或_writev接口，如下所示(只使用两种方法)：
/*classMyWritableextendsWritable{
constructor(){
super();
}
_write(chunk,encoding,callback){
process.stdout.write(chunk);
callback();
}
}
constmyWritable=newMyWritable();*/
constmyWritable=newWritable({
write(chunk,encoding,callback){
process.stdout.write(chunk);
callback();
}
});
myWritable.on('finish',()=>{
process.stdout.write('done');
})
myWritable.write('a');
myWritable.write('b');
myWritable.write('c');
myWritable.end();
Duplex流实现
实现Duplex流，需要继承Duplex类，并实现_read和_write接口，如下所示：
classMyDuplexextendsDuplex{
constructor(){
super();
this.source=[];
}
_read(){
if(!this.source.length){
this.push(null);
}else{
this.push(this.source.shift());
}
}
_write(chunk,encoding,cb){
this.source.push(chunk);
cb();
}
}
constmyDuplex=newMyDuplex();
myDuplex.on('finish',()=>{
process.stdout.write('writedone.')
});
myDuplex.on('end',()=>{
process.stdout.write('readdone.')
});
myDuplex.write('\na\n');
myDuplex.write('c\n');
myDuplex.end('b\n');
myDuplex.pipe(process.stdout);
上面的代码实现了_read()方法，可作为可读流来使用，同时实现了_write()方法，又可作为可写流来使用。
Transform流实现
实现Transform流，需要继承Transform类，并实现_transform接口，如下所示：
classMyTransformextendsTransform{
constructor(){
super();
}
_transform(chunk,encoding,callback){
chunk=(chunk+'').toUpperCase();
callback(null,chunk);
}
}
constmyTransform=newMyTransform();
myTransform.write('helloworld!');
myTransform.end();
myTransform.pipe(process.stdout);
上面代码中的_transform()方法，其第一个参数，要么为error，要么为null，第二个参数将被自动转发给readable.push()方法，因此该方法也可以使用如下写法：
_transform(chunk,encoding,callback){
chunk=(chunk+'').toUpperCase()
this.push(chunk)
callback();
}
ObjectMode流实现
我们知道流中的数据默认都是Buffer类型，可读流的数据进入流中便被转换成buffer，然后被消耗，可写流写入数据时，底层调用也将其转化为buffer。但将构造函数的objectMode选择设置为true，便可产生原样的数据，如下所示：
constrs=Readable();
rs.push('a');
rs.push('b');
rs.push(null);
rs.on('data',(chunk)=>{console.log(chunk);});//<Buffer61>与<Buffer62>
constrs1=Readable({objectMode:!0});
rs1.push('a');
rs1.push('b');
rs1.push(null);
rs1.on('data',(chunk)=>{console.log(chunk);});//a与b
下面利用Transform流实现一个简单的CSS压缩工具，如下所示：
functionminify(src,dest){
consttransform=newTransform({
transform(chunk,encoding,cb){
cb(null,(chunk.toString()).replace(/[\s\r\n\t]/g,''));
}
});
fs.createReadStream(src,{encoding:'utf8'}).pipe(transform).pipe(fs.createWriteStream(dest));
}
minify('./reset.css','./reset.min.css');
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。