首页 > 前端开发 > Node.js开发入门—Stream用法详解

Node.js开发入门—Stream用法详解

Stream是Node.js中非常重要的一个模块,离开它,基本上你就废了,除了写个HelloWorld,其它什么也干不成……...

Stream是Node.js中非常重要的一个模块,应用广泛。一个流是一个具备了可读、可写或既可读又可写能力的接口,通过这些接口,我们可以和磁盘文件、套接字、HTTP请求来交互,实现数据从一个地方流动到另一个地方的功能。

所有的流都实现了EventEmitter的接口,具备事件能力,通过发射事件来反馈流的状态。比如有错误发生时会发射“error”事件,有数据可被读取时发射“data”事件。这样我们就可以注册监听器来处理某个事件,达到我们的目的。

Node.js定义了Readable、Writable、Duplex、Transform四种流,Node.js有各种各样的模块,分别实现了这些流,我们挑出来一一看一看他们的用法。当然,我们也可以实现自己的流,可以参考Stream的文档或我们即将提到的这些Node.js里的实现。

Readable

Readable流提供了一种将外部来源(比如文件、套接字等)的数据读入到应用程序的机制。

可读的流有两种模式:流动模式和暂停模式。流动模式下,数据会自动从来源流出,跟不老泉似的,直到来源的数据耗尽。暂停模式下,你得通过stream.read()主动去要数据,你要了它才从来源读,你不要它就在那儿耗着等你。

可读流在创建时都是暂停模式。暂停模式和流动模式可以互相转换。

要从暂停模式切换到流动模式,有下面三种办法:

  • 给“data”事件关联了一个处理器
  • 显式调用resume()
  • 调用pipe()将可读流桥接到一个可写流上

要从流动模式切换到暂停模式,有两种途径:

  • 如果这个可读的流没有桥接可写流组成管道,直接调用pause()
  • 如果这个可读的流与若干可写流组成了管道,需要移除与“data”事件关联的所有处理器,并且调用unpipe()方法断开所有管道。

需要注意的是,出于向后兼容的原因,移除“data”事件的处理器,可读流并不会自动从流动模式转换到暂停模式;还有,对于已组成管道的可读流,调用pause也不能保证这个流会转换到暂停模式。

Readable流的一些常见实例如下:

  • 客户端的HTTP响应
  • 服务端的HTTP请求
  • fs读取流
  • zlib流
  • crypto(加密)流
  • TCP套接字
  • 子进程的stdout和stderr
  • process.stdin

Readable流提供了以下事件:

  • readable:在数据块可以从流中读取的时候发出。它对应的处理器没有参数,可以在处理器里调用read([size])方法读取数据。
  • data:有数据可读时发出。它对应的处理器有一个参数,代表数据。如果你只想快快地读取一个流的数据,给data关联一个处理器是最方便的办法。处理器的参数是Buffer对象,如果你调用了Readable的setEncoding(encoding)方法,处理器的参数就是String对象。
  • end:当数据被读完时发出。对应的处理器没有参数。
  • close:当底层的资源,如文件,已关闭时发出。不是所有的Readable流都会发出这个事件。对应的处理器没有参数。
  • error:当在接收数据中出现错误时发出。对应的处理器参数是Error的实例,它的message属性描述了错误原因,stack属性保存了发生错误时的堆栈信息。

Readable还提供了一些函数,我们可以用它们读取或操作流:

  • read([size]):如果你给read方法传递了一个大小作为参数,那它会返回指定数量的数据,如果数据不足,就会返回null。如果你不给read方法传参,它会返回内部缓冲区里的所有数据,如果没有数据,会返回null,此时有可能说明遇到了文件末尾。read返回的数据可能是Buffer对象,也可能是String对象。
  • setEncoding(encoding):给流设置一个编码格式,用于解码读到的数据。调用此方法后,read([size])方法返回String对象。
  • pause():暂停可读流,不再发出data事件
  • resume():恢复可读流,继续发出data事件
  • pipe(destination,[options]):把这个可读流的输出传递给destination指定的Writable流,两个流组成一个管道。options是一个JS对象,这个对象有一个布尔类型的end属性,默认值为true,当end为true时,Readable结束时自动结束Writable。注意,我们可以把一个Readable与若干Writable连在一起,组成多个管道,每一个Writable都能得到同样的数据。这个方法返回destination,如果destination本身又是Readable流,就可以级联调用pipe(比如我们在使用gzip压缩、解压缩时就会这样,马上会讲到)。
  • unpipe([destination]):端口与指定destination的管道。不传递destination时,断开与这个可读流连在一起的所有管道。

好吧,大概就这些了,我们来举一个简单的使用Readable的例子。以fs模块为例吧。

fs.ReadStream实现了stream.Readable,另外还提供了一个“open”事件,你可以给这个事件关联处理器,处理器的参数是文件描述符(一个整型数)。

fs.createReadStream(path[, options])用来打开一个可读的文件流,它返回一个fs.ReadStream对象。path参数指定文件的路径,可选的options是一个JS对象,可以指定一些选项,类似下面这样:

{ flags: 'r',   encoding: 'utf8',   fd: null,   mode: 0666,   autoClose: true } 

options的flags属性指定用什么模式打开文件,’w’代表写,’r’代表读,类似的还有’r+’、’w+’、’a’等,与Linux下的open函数接受的读写模式类似。encoding指定打开文件时使用编码格式,默认就是“utf8”,你还可以为它指定”ascii”或”base64”。fd属性默认为null,当你指定了这个属性时,createReadableStream会根据传入的fd创建一个流,忽略path。另外你要是想读取一个文件的特定区域,可以配置start、end属性,指定起始和结束(包含在内)的字节偏移。autoClose属性为true(默认行为)时,当发生错误或文件读取结束时会自动关闭文件描述符。

OK,背景差不多了,可以上代码了,readable.js文件内容如下:

var fs = require('fs');  var readable = fs.createReadStream('readable.js',{   flags: 'r',   encoding: 'utf8',   autoClose: true,   mode: 0666, });  readable.on('open', function(fd){   console.log('file was opened, fd - ', fd); });  readable.on('readable', function(){   console.log('received readable'); });  readable.on('data', function(chunk){   console.log('read %d bytes: %s', chunk.length, chunk); });  readable.on('end', function(){   console.log('read end'); });  readable.on('close', function(){   console.log('file was closed.'); });  readable.on('error', function(err){   console.log('error occured: %s', err.message); }); 

示例代码把readable.js的内容读取出来,关联了各种事件,演示了读取文件的一般用法。

Writable

Writable流提供了一个接口,用来把数据写入到目的设备(或内存)中。Writable流的一些常见实例:

  • 客户端的HTTP请求
  • 服务器的HTTP响应
  • fs写入流
  • zlib流
  • crypto(加密)流
  • TCP套结字
  • 子进程的stdin
  • process.stdout和process.stderr

Writable流的write(chunk[,encoding] [,callback])方法可以把数据写入流中。其中,chunk是待写入的数据,是Buffer或String对象。这个参数是必须的,其它参数都是可选的。如果chunk是String对象,encoding可以用来指定字符串的编码格式,write会根据编码格式将chunk解码成字节流再来写入。callback是数据完全刷新到流中时会执行的回调函数。write方法返回布尔值,当数据被完全处理后返回true(不一定是完全写入设备哦)。

Writable流的end([chunk] [,encoding] [,callback])方法可以用来结束一个可写流。它的三个参数都是可选的。chunk和encoding的含义与write方法类似。callback是一个可选的回调,当你提供它时,它会被关联到Writable的finish事件上,这样当finish事件发射时它就会被调用。

Writable还有setDefaultEncoding等方法,具体可以参考在线文档。

现在我们来看看Writable公开的事件:

  • finish: 在end()被调用、所有数据都已被写入底层设备后发射。对应的处理器函数没有参数。
  • pipe: 当你在Readable流上调用pipe()方法时,Writable流会发射这个事件,对应的处理器函数有一个参数,类型是Readable,指向与它连接的那个Readable流。
  • unpipe: 当你在Readable流上调用unpipe()方法时,Writable流会发射这个事件,对应的处理器函数有一个参数,类型是Readable,指向与刚与它断开连接的那个Readable流。
  • error: 出错时发射,对应的处理器函数的参数是Error对象。

OK,让我们来举两个小例子。一个是fs的,一个是socket的。

fs.createWriteStream(path[,options])用来创建一个可写的文件流,它返回fs.WriteStream对象。第一个参数path是路径,第二个参数options是JS对象,是可选的,指定创建文件时的选项,类似:

{ flags: 'w',   defaultEncoding: 'utf8',   fd: null,   mode: 0666 } 

defaultEncoding指定默认的文本编码。前面讲fs.createReadStream时提到了。

writeFile.js内容如下:

var fs = require('fs');  var writable = fs.createWriteStream('example.txt',{   flags: 'w',   defaultEncoding: 'utf8',   mode: 0666, });  writable.on('finish', function(){   console.log('write finished');   process.exit(0); });  writable.on('error', function(err){   console.log('write error - %s', err.message); });  writable.write('My name is 火云邪神', 'utf8');  writable.end(); 

很简单的一个示例,注意writeFile.js的文件编码格式要是UTF8哦。

下面看一个使用TCP套接字的示例,echoServer2.js内容如下:

var net = require("net");  var server = net.createServer(function(sock){   sock.setEncoding('utf8');   sock.on('pipe', function(src){     console.log('piped');   });   sock.on('error', function(err){     console.log('error - %s', err.message);   });   sock.pipe(sock); });  server.maxConnections = 10; server.listen(7, function(){   console.log('echo server bound at port - 7'); }); 

上面的echoServer功能和我们之前在Node.js开发入门——套接字(socket)编程中的echoServer一样。不同的是,这里使用了pipe方法,而那个版本监听data事件,调用write方法将收到的数据回写给客户端。

sock.Socket是Duplex流,既实现了Readable又实现了Writable,所以,sock.pipe(sock)是正确的调用。

常见的Duplex流有:

  • TCP socket
  • zlib
  • crypto

Duplex是Readable和Writable的合体。

Transform

Transform扩展了Duplex流,它会修改你使用Writable接口写入的数据,当你用Readable接口来读时,数据已经发生了变化。

比较常见的Transform流有:

  • zlib
  • crypto

好啦,我们举一个简单的示例,使用zlib模块来压缩和解压缩。示例文件是zlibFile.js,内容如下:

var zlib = require("zlib"); var gzip = zlib.createGzip(); var fs = require('fs'); var inFile = fs.createReadStream('readable.js'); var outGzip = fs.createWriteStream('readable.gz'); //inFile - Readable //gzip - Transform(Readable && Writable) //outFile - Writable inFile.pipe(gzip).pipe(outGzip);  setTimeout(function(){   var gunzip = zlib.createUnzip({flush: zlib.Z_FULL_FLUSH});   var inGzip = fs.createReadStream('readable.gz');   var outFile = fs.createWriteStream('readable.unzipped');   inGzip.pipe(gunzip).pipe(outFile); }, 5000); 

上面的示例比较简单,使用了zlib模块,文档在这里:https://nodejs.org/api/zlib.html

接下来我们来实现一个Transform流,把输入数据中的小写字母转换为大写字母。我们代码在upperTransform.js里,内容如下:

var fs = require('fs'); var util = require('util'); var stream = require('stream'); util.inherits(UpperTransform, stream.Transform);  function UpperTransform(opt){   stream.Transform.call(this, opt); }  UpperTransform.prototype._transform = function(chunk, encoding, callback){   var data = new Buffer(chunk.length);   var str = chunk.toString('utf8');   for(var i = 0, offset=0; i < str.length; i++){     if(/^[a-z]+$/.test(str[i])){       offset += data.write(str[i].toUpperCase(), offset);     }else{       offset += data.write(str[i], offset);     }   }   this.push(data);   callback(); }  UpperTransform.prototype._flush = function(cb){   cb(); }  var upper = new UpperTransform(); var inFile = fs.createReadStream('example.txt'); inFile.setEncoding('utf8'); var outFile = fs.createWriteStream('exampleUpper.txt',{defaultEncoding: 'utf8'});  inFile.pipe(upper).pipe(outFile); 

为了实现自定义的Transform,需要先继承Transform流的功能。实现这一点最简单的办法就是使用util模块的inherits()方法,然后在你的构造器里调用使用call方法把父对象应用到当前对象上。代码就是下面这部分:

util.inherits(UpperTransform, stream.Transform); function UpperTransform(opt){   stream.Transform.call(this, opt); } 

继承了stream.Transform之后,实现_transform和_flush即可。在_transform里,我们先创建了一个缓冲区,然后把传入的数据(chunk)转换成字符串(写死为utf8了),接着遍历字符串,遇见小写字母就转换一下,写入创建的缓冲区里,完成转换后,调用push方法,把转换后的数据加到内部的数据队列中。

其它的就比较简单了。注意,作为示例,我们只转换utf8编码的文本文件。


其它文章:

推荐阅读