ITPub博客

首页 > 应用开发 > Node.js > 好程序员web前端分享Nodejs学习笔记之Stream模块 

好程序员web前端分享Nodejs学习笔记之Stream模块 

原创 Node.js 作者:好程序员IT 时间:2019-05-17 14:23:17 0 删除 编辑

好程序员 web前端分享 Nodejs学习笔记之Stream模块
  一,开篇分析

  流是一个抽象接口,被 Node 中的很多对象所实现。比如对一个 HTTP 服务器的请求是一个流,stdout 也是一个流。流是可读,可写或兼具两者的。

  最早接触 Stream是从早期的unix开始的, 数十年的实践证明Stream 思想可以很简单的开发出一些庞大的系统。

  在 unix里,Stream是通过 "|" 实现的。在node中,作为内置的stream模块,很多核心模块和三方模块都使用到。

  和 unix一样,node stream主要的操作也是.pipe(),使用者可以使用反压力机制来控制读和写的平衡。

  Stream 可以为开发者提供可以重复使用统一的接口,通过抽象的Stream接口来控制Stream之间的读写平衡。

  一个 TCP连接既是可读流,又是可写流,而Http连接则不同,一个http request对象是可读流,而http response对象则是可写流。

  流的传输过程默认是以 buffer的形式传输的,除非你给他设置其他编码形式,以下是一个例子:

1.  <p><font size="3">

2.  </font></p>

3.  <p><font size="3">  var http = require('http') ;</font></p>

4.  <p><font size="3">  var server = http.createServer(function(req,res){</font></p>

5.  <p><font size="3">  res.writeHeader(200, {'Content-Type': 'text/plain'}) ;</font></p>

6.  <p><font size="3">  res.end("Hello,大熊!") ;</font></p>

7.  <p><font size="3">  }) ;</font></p>

8.  <p><font size="3">  server.listen(8888) ;</font></p>

9.  <p><font size="3">  console.log("http server running on port 8888 ...") ;</font></p>


  运行后会有乱码出现,原因就是没有设置指定的字符集,比如: “utf-8” 。

  修改一下就好:

   var http = require('http') ;

1. 

2.  <p><font size="3">  var server = http.createServer(function(req,res){</font></p>

3.  <p><font size="3">  res.writeHeader(200,{</font></p>

4.  <p><font size="3">  'Content-Type' : 'text/plain;charset=utf-8' // 添加charset=utf-8</font></p>

5.  <p><font size="3">  }) ;</font></p>

6.  <p><font size="3">  res.end("Hello,大熊!") ;</font></p>

7.  <p><font size="3">  }) ;</font></p>

8.  <p><font size="3">  server.listen(8888) ;</font></p>

9.  <p><font size="3">  console.log("http server running on port 8888 ...") ;</font></p>


  运行结果:

  为什么使用 Stream

  node中的I/O是异步的,因此对磁盘和网络的读写需要通过回调函数来读取数据,下面是一个文件下载例子

   上代码:

1. 

2.  <p><font size="3">  var http = require('http') ;</font></p>

3.  <p><font size="3">  var fs = require('fs') ;</font></p>

4.  <p><font size="3">  var server = http.createServer(function (req, res) {</font></p>

5.  <p><font size="3">  fs.readFile(__dirname + '/data.txt', function (err, data) {</font></p>

6.  <p><font size="3">  res.end(data);</font></p>

7.  <p><font size="3">  }) ;</font></p>

8.  <p><font size="3">  }) ;</font></p>

9.  <p><font size="3">  server.listen(8888) ;</font></p>


  代码可以实现需要的功能,但是服务在发送文件数据之前需要缓存整个文件数据到内存,如果 "data.txt"文件很

  大并且并发量很大的话,会浪费很多内存。因为用户需要等到整个文件缓存到内存才能接受的文件数据,这样导致

  用户体验相当不好。不过还好 (req,res)两个参数都是Stream,这样我们可以用fs.createReadStream()代替fs.readFile()。如下:

   var http = require('http') ;

1. 

2.  <p><font size="3">  var fs = require('fs') ;</font></p>

3.  <p><font size="3">  var server = http.createServer(function (req, res) {</font></p>

4.  <p><font size="3">  var stream = fs.createReadStream(__dirname + '/data.txt') ;</font></p>

5.  <p><font size="3">  stream.pipe(res) ;</font></p>

6.  <p><font size="3">  }) ;</font></p>

7.  <p><font size="3">  server.listen(8888) ;</font></p>


  .pipe()方法监听fs.createReadStream()的'data' 和'end'事件,这样"data.txt"文件就不需要缓存整

  个文件,当客户端连接完成之后马上可以发送一个数据块到客户端。使用 .pipe()另一个好处是可以解决当客户

  端延迟非常大时导致的读写不平衡问题。

  有五种基本的 Stream:readable,writable,transform,duplex,and "classic” 。(具体使用请自己查阅api)

  二,实例引入

  当内存中无法一次装下需要处理的数据时,或者一边读取一边处理更加高效时,我们就需要用到数据流。 NodeJS中通过各种Stream来提供对数据流的操作。

  以大文件拷贝程序为例,我们可以为数据源创建一个只读数据流,示例如下:

   var rs = fs.createReadStream(pathname);

1. 

2.  <p><font size="3">  rs.on('data', function (chunk) {</font></p>

3.  <p><font size="3">  doSomething(chunk) ; // 具体细节自己任意发挥</font></p>

4.  <p><font size="3">  });</font></p>

5.  <p><font size="3">  rs.on('end', function () {</font></p>

6.  <p><font size="3">  cleanUp() ;</font></p>

7.  <p><font size="3">  }) ;</font></p>


  代码中 data事件会源源不断地被触发,不管doSomething函数是否处理得过来。代码可以继续做如下改造,以解决这个问题。

1.  <p><font size="3">

2.  </font></p>

3.  <p><font size="3">  var rs = fs.createReadStream(src) ;</font></p>

4.  <p><font size="3">  rs.on('data', function (chunk) {</font></p>

5.  <p><font size="3">  rs.pause() ;</font></p>

6.  <p><font size="3">  doSomething(chunk, function () {</font></p>

7.  <p><font size="3">  rs.resume() ;</font></p>

8.  <p><font size="3">  }) ;</font></p>

9.  <p><font size="3">  }) ;</font></p>

10.  <p><font size="3">  rs.on('end', function () {</font></p>

11.  <p><font size="3">  cleanUp();</font></p>

12.  <p><font size="3">  }) ;</font></p>


  给 doSomething函数加上了回调,因此我们可以在处理数据前暂停数据读取,并在处理数据后继续读取数据。

  此外,我们也可以为数据目标创建一个只写数据流,如下:
   var rs = fs.createReadStream(src) ;

1. 

2.  <p><font size="3">  var ws = fs.createWriteStream(dst) ;</font></p>

3.  <p><font size="3">  rs.on('data', function (chunk) {</font></p>

4.  <p><font size="3">  ws.write(chunk);</font></p>

5.  <p><font size="3">  }) ;</font></p>

6.  <p><font size="3">  rs.on('end', function () {</font></p>

7.  <p><font size="3">  ws.end();</font></p>

8.  <p><font size="3">  }) ;</font></p>


  doSomething换成了往只写数据流里写入数据后,以上代码看起来就像是一个文件拷贝程序了。但是以上代码存在上边提到的问题,如果写入速度跟不上读取速度的话,只写数据流内部的缓存会爆仓。我们可以根据.write方法的返回值来判断传入的数据是写入目标了,还是临时放在了缓存了,并根据drain事件来判断什么时候只写数据流已经将缓存中的数据写入目标,可以传入下一个待写数据了。因此代码如下:

   var rs = fs.createReadStream(src) ;

1. 

2.  <p><font size="3">  var ws = fs.createWriteStream(dst) ;</font></p>

3.  <p><font size="3">  rs.on('data', function (chunk) {</font></p>

4.  <p><font size="3">  if (ws.write(chunk) === false) {</font></p>

5.  <p><font size="3">  rs.pause() ;</font></p>

6.  <p><font size="3">  }</font></p>

7.  <p><font size="3">  }) ;</font></p>

8.  <p><font size="3">  rs.on('end', function () {</font></p>

9.  <p><font size="3">  ws.end();</font></p>

10.  <p><font size="3">  });</font></p>

11.  <p><font size="3">  ws.on('drain', function () {</font></p>

12.  <p><font size="3">  rs.resume();</font></p>

13.  <p><font size="3">  }) ;</font></p>


  最终实现了数据从只读数据流到只写数据流的搬运,并包括了防爆仓控制。因为这种使用场景很多,例如上边的大文件拷贝程序, NodeJS直接提供了.pipe方法来做这件事情,其内部实现方式与上边的代码类似。

  下面是一个更加完整的复制文件的过程:

   var fs = require('fs'),

1. 

2.  <p><font size="3">  path = require('path'),</font></p>

3.  <p><font size="3">  out = process.stdout;</font></p>

4.  <p><font size="3">  var filePath = '/bb/bigbear.mkv';</font></p>

5.  <p><font size="3">  var readStream = fs.createReadStream(filePath);</font></p>

6.  <p><font size="3">  var writeStream = fs.createWriteStream('file.mkv');</font></p>

7.  <p><font size="3">  var stat = fs.statSync(filePath);</font></p>

8.  <p><font size="3">  var totalSize = stat.size;</font></p>

9.  <p><font size="3">  var passedLength = 0;</font></p>

10.  <p><font size="3">  var lastSize = 0;</font></p>

11.  <p><font size="3">  var startTime = Date.now();</font></p>

12.  <p><font size="3">  readStream.on('data', function(chunk) {</font></p>

13.  <p><font size="3">  passedLength += chunk.length;</font></p>

14.  <p><font size="3">  if (writeStream.write(chunk) === false) {</font></p>

15.  <p><font size="3">  readStream.pause();</font></p>

16.  <p><font size="3">  }</font></p>

17.  <p><font size="3">  });</font></p>

18.  <p><font size="3">  readStream.on('end', function() {</font></p>

19.  <p><font size="3">  writeStream.end();</font></p>

20.  <p><font size="3">  });</font></p>

21.  <p><font size="3">  writeStream.on('drain', function() {</font></p>

22.  <p><font size="3">  readStream.resume();</font></p>

23.  <p><font size="3">  });</font></p>

24.  <p><font size="3">  setTimeout(function show() {</font></p>

25.  <p><font size="3">  var percent = Math.ceil((passedLength / totalSize) * 100);</font></p>

26.  <p><font size="3">  var size = Math.ceil(passedLength / 1000000);</font></p>

27.  <p><font size="3">  var diff = size - lastSize;</font></p>

28.  <p><font size="3">  lastSize = size;</font></p>

29.  <p><font size="3">  out.clearLine();</font></p>

30.  <p><font size="3">  out.cursorTo(0);</font></p>

31.  <p><font size="3">  out.write('已完成' + size + 'MB, ' + percent + '%, 速度:' + diff * 2 + 

32.  'MB/s');</font></p>

33.  <p><font size="3">  if (passedLength < totalSize) {</font></p>

34.  <p><font size="3">  setTimeout(show, 500);</font></p>

35.  <p><font size="3">  } else {</font></p>

36.  <p><font size="3">  var endTime = Date.now();</font></p>

37.  <p><font size="3">  console.log();</font></p>

38.  <p><font size="3">  console.log('共用时:' + (endTime - startTime) / 1000 + '秒。');</font></p>

39.  <p><font size="3">  }</font></p>

40.  <p><font size="3">  }, 500);</font></p>


  可以把上面的代码保存为 "copy.js" 试验一下我们添加了一个递归的 setTimeout (或者直接使用setInterval)来做一个旁观者,

  每 500ms观察一次完成进度,并把已完成的大小、百分比和复制速度一并写到控制台上,当复制完成时,计算总的耗费时间。

  三,总结一下
  (1),理解Stream概念。
  (2),熟练使用相关Stream的api
  (3),注意细节的把控,比如:大文件的拷贝,采用的使用 “chunk data” 的形式进行分片处理。
   (4),pipe的使用
  (5),再次强调一个概念:一个TCP连接既是可读流,又是可写流,而Http连接则不同,一个http request对象是可读流,而http response对象则是可写流。


来自 “ ITPUB博客 ” ,链接:http://blog.itpub.net/69913892/viewspace-2644730/,如需转载,请注明出处,否则将追究法律责任。

请登录后发表评论 登录
全部评论

注册时间:2019-03-20

  • 博文量
    182
  • 访问量
    142486