源码网商城,靠谱的源码在线交易网站 我的订单 购物车 帮助

源码网商城

浅谈Node.js:理解stream

  • 时间:2022-06-07 06:45 编辑: 来源: 阅读:
  • 扫一扫,手机访问
摘要:浅谈Node.js:理解stream
Stream在node.js中是一个抽象的接口,基于EventEmitter,也是一种Buffer的高级封装,用来处理流数据。流模块便是提供各种API让我们可以很简单的使用Stream。 流分为四种类型,如下所示: [list] [*]Readable,可读流[/*] [*]Writable,可写流[/*] [*]Duplex,读写流[/*] [*]Transform,扩展的Duplex,可修改写入的数据[/*] [/list] [b]1、Readable可读流[/b] 通过stream.Readable可创建一个可读流,它有两种模式:暂停和流动。 在流动模式下,将自动从下游系统读取数据并使用data事件输出;暂停模式下,必须显示调用[code]stream.read()[/code]方法读取数据,并触发data事件。 所有的可读流最开始都是暂停模式,可以通过以下方法切换到流动模式: [list] [*]监听'data'事件[/*] [*]调用[code]stream.resume()[/code]方法[/*] [*]调用[code]stream.pipe()[/code]方法将数据输出到一个可写流Writable[/*] [/list] 同样地,也可以切换到暂停模式,有两种方法: [list] [*]如果没有设置pipe目标,调用[code]stream.pause()[/code]方法即可。[/*] [*]如果设置了pipe目标,则需要移除所有的data监听和调用[code]stream.unpipe()[/code]方法[/*] [/list] 在Readable对象中有一个[code]_readableSate[/code]的对象,通过该对象可以得知流当前处于什么模式,如下所示: [list] [*]readable._readableState.flowing = null,没有数据消费者,流不产生数据[/*] [*]readable._readableState.flowing = true,处于流动模式[/*] [*]readable._readableState.flowing = false,处于暂停模式[/*] [/list] [b]为什么使用流取数据[/b] 对于小文件,使用[code]fs.readFile()[/code]方法读取数据更方便,但需要读取大文件的时候,比如几G大小的文件,使用该方法将消耗大量的内存,甚至使程序崩溃。这种情况下,使用流来处理是更合适的,采用分段读取,便不会造成内存的'爆仓'问题。 [b]data事件[/b] 在stream提供数据块给消费者时触发,有可能是切换到流动模式的时候,也有可能是调用[code]readable.read()[/code]方法且有有效数据块的时候,使用如下所示:
const fs = require('fs');

const rs = fs.createReadStream('./appbak.js');
var chunkArr = [],
  chunkLen = 0;
rs.on('data',(chunk)=>{
  chunkArr.push(chunk);
  chunkLen+=chunk.length;
});
rs.on('end',(chunk)=>{
  console.log(Buffer.concat(chunkArr,chunkLen).toString());
});

[b]readable事件[/b] 当流中有可用数据能被读取时触发,分为两种,新的可用的数据和到达流的末尾,前者[code]stream.read()[/code]方法返回可用数据,后者返回null,如下所示:
const rs = fs.createReadStream('./appbak.js');
var chunkArr = [],
  chunkLen = 0;

rs.on('readable',()=>{
  var chunk = null;
  //这里需要判断是否到了流的末尾
  if((chunk = rs.read()) !== null){
    chunkArr.push(chunk);
    chunkLen+=chunk.length;
  }
});
rs.on('end',(chunk)=>{
  console.log(Buffer.concat(chunkArr,chunkLen).toString());
});

[b]pause和resume方法[/b] [code]stream.pause()[/code]方法让流进入暂停模式,并停止'data'事件触发,[code]stream.resume()[/code]方法使流进入流动模式,并恢复'data'事件触发,也可以用来消费所有数据,如下所示:
const rs = 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(`数据接收完毕`);
});
[b]pipe(destination[, options])方法[/b] [code]pipe()[/code]方法绑定一个可写流到可读流上,并自动切换到流动模式,将所有数据输出到可写流,以及做好了数据流的管理,不会发生数据丢失的问题,使用如下所示:
const rs = fs.createReadStream('./app.js');
rs.pipe(process.stdout);
以上介绍了多种可读流的数据消费的方法,但对于一个可读流,最好只选择其中的一种,推荐使用[code]pipe()[/code]方法。 [b]2、Writable可写流[/b] 所有的可写流都是基于[code]stream.Writable[/code]类创建的,创建之后便可将数据写入该流中。 write(chunk[, encoding][, callback])方法 [code]write()[/code]方法向可写流中写入数据,参数含义: [list] [*]chunk,字符串或buffer[/*] [*]encoding,若chunk为字符串,则是chunk的编码[/*] [*]callback,当前chunk数据写入磁盘时的回调函数[/*] [/list] 该方法的返回值为布尔值,如果为false,则表示需要写入的数据块被缓存并且此时缓存的大小超出highWaterMark阀值,否则为true。  使用如下所示:
const ws = fs.createWriteStream('./test.txt');
ws.write('nihao','utf8',()=>{process.stdout.write('this chunk is flushed.');});
ws.end('done.')
[b]背压机制[/b] 如果可写流的写入速度跟不上可读流的读取速度,write方法添加的数据将被缓存,逐渐增多,导致占用大量内存。我们希望的是消耗一个数据,再去读取一个数据,这样内存就维持在一个水平上。如何做到这一点?可以利用write方法的返回值来判断可写流的缓存状态和'drain'事件,及时切换可读流的模式,如下所示:
function copy(src,dest){
  src = path.resolve(src);
  dest = path.resolve(dest);
  const rs = fs.createReadStream(src);
  const ws = fs.createWriteStream(dest);
  console.log('正在复制中...');
  const stime = +new Date();
  rs.on('data',(chunk)=>{
    if(null === ws.write(chunk)){
      rs.pause();
    }
  });
  ws.on('drain',()=>{
    rs.resume();
  });
  rs.on('end',()=>{
    const etime = +new Date();
    console.log(`已完成,用时:${(etime-stime)/1000}秒`);
    ws.end();
  });
  function calcProgress(){
    
  }
}
copy('./CSS权威指南 第3版.pdf','./javascript.pdf');

[b]drain事件[/b] 如果[code]Writable.write()[/code]方法返回false,则drain事件将会被触发,上面的背压机制已经使用了该事件。 [b]finish事件[/b] 在调用[code]stream.end()[/code]方法之后且所有缓存区的数据都被写入到下游系统,就会触发该事件,如下所示:
const ws = fs.createWriteStream('./alphabet.txt');
const alphabetStr = 'abcdefghijklmnopqrstuvwxyz';
ws.on('finish',()=>{
  console.log('done.');
});
for(let letter of alphabetStr.split()){
  ws.write(letter);
}
ws.end();//必须调用
end([chunk][, encoding][, callback])方法 [code]end()[/code]方法被调用之后,便不能再调用[code]stream.write()[/code]方法写入数据,负责将抛出错误。 [b]3、Duplex读写流[/b] Duplex流同时实现了Readable与Writable类的接口,既是可读流,也是可写流。例如'zlib streams'、'crypto streams'、'TCP sockets'等都是Duplex流。 [b]4、Transform流[/b] Duplex流的扩展,区别在于,Transform流自动将写入端的数据变换后添加到可读端。例如:'zlib streams'、'crypto streams'等都是Transform流。 [b]5、四种流的实现[/b] [code]stream[/code]模块提供的API可以让我们很简单的实现流,该模块使用[code]require('stream')[/code]引用,我们只要继承四种流中的一个基类[code](stream.Writable, stream.Readable, stream.Duplex, or stream.Transform)[/code],然后实现它的接口就可以了,需要实现的接口如下所示: | Use-case | Class | Method(s) to implement |  | ------------- |-------------| -----|  | Reading only | Readable | _read |  | Writing only | Writable | _write, _writev |  | Reading and writing | Duplex | _read, _write, _writev |  | Operate on written data, then read the result | Transform | _transform, _flush | [b]Readable流实现[/b] 如上所示,我们只要继承Readable类并实现_read接口即可,,如下所示:
const Readable = require('stream').Readable;
const util = require('util');
const alphabetArr = 'abcdefghijklmnopqrstuvwxyz'.split();
/*function AbReadable(){
  if(!this instanceof AbReadable){
    return new AbReadable();
  }
  Readable.call(this);
}
util.inherits(AbReadable,Readable);
AbReadable.prototype._read = function(){
  if(!alphabetArr.length){
    this.push(null);
  }else{
    this.push(alphabetArr.shift());
  }
};

const abReadable = new AbReadable();
abReadable.pipe(process.stdout);*/

/*class AbReadable extends Readable{
  constructor(){
    super();
  }
  _read(){
    if(!alphabetArr.length){
      this.push(null);
    }else{
      this.push(alphabetArr.shift());
    }
  }
}
const abReadable = new AbReadable();
abReadable.pipe(process.stdout);*/

/*const abReadable = new Readable({
  read(){
    if(!alphabetArr.length){
      this.push(null);
    }else{
      this.push(alphabetArr.shift());
    }
  }
});
abReadable.pipe(process.stdout);*/

const abReadable = Readable();
abReadable._read = function(){
  if (!alphabetArr.length) {
    this.push(null);
  } else {
    this.push(alphabetArr.shift());
  }
}
abReadable.pipe(process.stdout);

以上代码使用了四种方法创建一个Readable可读流,必须实现[code]_read()[/code]方法,以及用到了[code]readable.push()[/code]方法,该方法的作用是将指定的数据添加到读取队列。 [b]Writable流实现 [/b] 我们只要继承Writable类并实现_write或_writev接口,如下所示(只使用两种方法):
/*class MyWritable extends Writable{
  constructor(){
    super();
  }
  _write(chunk,encoding,callback){
    process.stdout.write(chunk);
    callback();
  }
}
const myWritable = new MyWritable();*/
const myWritable = new Writable({
  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();

[b]Duplex流实现[/b] 实现Duplex流,需要继承Duplex类,并实现_read和_write接口,如下所示:
class MyDuplex extends Duplex{
  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();
  }
}

const myDuplex = new MyDuplex();
myDuplex.on('finish',()=>{
  process.stdout.write('write done.')
});
myDuplex.on('end',()=>{
  process.stdout.write('read done.')
});
myDuplex.write('nan');
myDuplex.write('cn');
myDuplex.end('bn');
myDuplex.pipe(process.stdout);

上面的代码实现了[code]_read()[/code]方法,可作为可读流来使用,同时实现了[code]_write()[/code]方法,又可作为可写流来使用。 [b]Transform流实现 [/b] 实现Transform流,需要继承Transform类,并实现_transform接口,如下所示:
class MyTransform extends Transform{
  constructor(){
    super();
  }
  _transform(chunk, encoding, callback){
    chunk = (chunk+'').toUpperCase();
    callback(null,chunk);
  }
}
const myTransform = new MyTransform();
myTransform.write('hello world!');
myTransform.end();
myTransform.pipe(process.stdout);
上面代码中的[code]_transform()[/code]方法,其第一个参数,要么为error,要么为null,第二个参数将被自动转发给[code]readable.push()[/code]方法,因此该方法也可以使用如下写法:
_transform(chunk, encoding, callback){
  chunk = (chunk+'').toUpperCase()
  this.push(chunk)
  callback();
}
[b]Object Mode流实现[/b] 我们知道流中的数据默认都是Buffer类型,可读流的数据进入流中便被转换成buffer,然后被消耗,可写流写入数据时,底层调用也将其转化为buffer。但将构造函数的objectMode选择设置为true,便可产生原样的数据,如下所示:
const rs = Readable();
rs.push('a');
rs.push('b');
rs.push(null);
rs.on('data',(chunk)=>{console.log(chunk);});//<Buffer 61>与<Buffer 62>

const rs1 = Readable({objectMode:!0});
rs1.push('a');
rs1.push('b');
rs1.push(null);
rs1.on('data',(chunk)=>{console.log(chunk);});//a与b

下面利用Transform流实现一个简单的CSS压缩工具,如下所示:
function minify(src,dest){
  const transform = new Transform({
    transform(chunk,encoding,cb){
      cb(null,(chunk.toString()).replace(/[srnt]/g,''));
    }
  });
  fs.createReadStream(src,{encoding:'utf8'}).pipe(transform).pipe(fs.createWriteStream(dest));
}
minify('./reset.css','./reset.min.css');
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持编程素材网。
  • 全部评论(0)
联系客服
客服电话:
400-000-3129
微信版

扫一扫进微信版
返回顶部