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

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编码的文本文件。


其它文章：

版权声明：本文为foruok原创文章，未经博主允许不得转载。

时间： 2024-10-27 12:43:08

Node.js开发入门—Stream用法详解的相关文章

Node.js开发入门—HTTP文件服务器

HelloWorld示例只有演示意义,这次我们来搞一个实际的例子:文件服务器.我们使用Node.js创建一个HTTP协议的文件服务器,你可以使用浏览器或其它下载工具到文件服务器上下载文件. 用Node.js实现的HTTP文件服务器,比我在Qt网络编程实战之HTTP服务器视频课程里用Qt实现的版本,复杂度要小太多太多了. 为了读取文件,我们会用到File System模块(名字是"fs"),Stream,我们还要分析URL,区别HTTP方法,还会用到EventEmitter. 文件服务器

Node.js开发入门—使用对话框ngDialog

做网站经常会遇到弹出对话框获取用户输入或弹出对话框让用户确认某个操作之类的情景,有一个基于AngularJS的扩展模块可以帮我们优雅地完成这类事情:ngDialog. ngDialog在github上提供了一个示例网页,演示了它的各种用法,在这里:https://github.com/likeastore/ngDialog/blob/master/example/index.html.ngDialog的github主页的readme也对常用的指令和服务做了较为详细的介绍,可以参考.我这篇就纯粹是

Node.js开发入门—Angular简单示例

在"使用AngularJS"中,我们提到了如何在Node.js项目中引入AngularJS,这次提供一个非常简单的示例,演示AngularJS里的指令.数据绑定.服务等内容. 我准备做Web后台管理系统,不同的管理员会有不同的权限,管理员登录后看到的菜单和他的权限有关,能看到什么,是动态生成的(类似RBAC).本文的示例从这个项目而来,当然,现在还是最简单的. 如果没有特别说明,后面我们用到的示例都使用express generator生成. Angular小demo 先搞起来吧. 第

Node.js开发入门—使用http访问外部世界

Node.js的http模块,不但可以构建服务器,也可以作为客户端类库来访问别的服务器.关键就在两个方法: http.request(options[,callback]) http.get(path[,callback]) 除了http,还会用到FileSystem模块和Stream中的stream.Readable和stream.Writable. 先来大概介绍一下相关API吧. API解释 http.request()方法接受一个options参数,这个参数可以是对象,用来指明你要访问的网

Node.js开发入门—套接字(socket)编程

Node.js的net模块提供了socket编程接口,方便我们利用较为底层的套接字接口来实现应用协议.这次我们看一个简单的回显服务器示例,包括服务端和客户端的代码. 使用JavaScript也可以进行套接字编程,哈哈,这酸爽! 代码 分服务器和客户端两部分来说吧. echoServer代码分析 echoServer.js: var net = require("net"); // server is an instance of net.Server // sock is an ins

Node.js开发入门—notepad++ for Node.js

对于Node.js开发,论IDE的话,Webstorm是不二的选择,但它是收费的(可免费使用30天).一开始,我们先将就一下,使用notepad++来编写Node.js应用.这样做还有一大好处:没有关于Node.js的代码高亮和自动补全,可以更好地敦促我们使用在线API文档记忆各种类库API.死磕自己吧,enjoy it. notepad++的安装与配置 到"https://notepad-plus-plus.org/download/v6.8.2.html"这里下载吧,6.8.2版本

Node.js开发入门—使用AngularJS内置服务

在上一篇,"AngularJS简单示例"中演示了一个非常简单的使用Angular的小demo,那篇已经太长,原本要介绍的一些内容只好单另开篇了.这些内容,就是如何使用Angular服务. 我们还是基于"AngularJS简单示例"中的示例来改造一下.新的示例,能从Node.js+Express构造的服务器上获取管理菜单.为了实现这个,需要做几部分改造: 服务器提供adminMenu的下载功能,需要修改app.js,处理路由 修改Angular实现的控制器x-cont

Node.js开发入门——MongoDB与Mongoose

为了保存网站的用户数据和业务数据,通常需要一个数据库.MongoDB和Node.js特别般配,因为MongoDB是基于文档的非关系型数据库,文档是按BSON(JSON的轻量化二进制格式)存储的,增删改查等管理数据库的命令和JavaScript语法很像.如果你在Node.js里访问MongoDB的数据,会有我们是一家人的感觉,特别亲切. 我也准备使用MongoDB来作为我的数据库. MongoDB使用集合(collection)和文档(document)来描述和存储数据,collection就相当

Node.js开发入门—使用cookie保持登录

这次来做一个网站登录的小例子,后面会用到.这个示例会用到Cookie.HTML表单.POST数据体(body)解析. 第一个版本,我们的用户数据就写死在js文件里.第二个版本会引入MongoDB来保存用户数据. 示例准备 1. 使用express创建应用 就下面的命令序列: express LoginDemo cd LoginDemo npm install 2. 登录页面 登录页面的jade模板为login.jade,内容如下: doctype html html head meta(char