文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

Nodejs学习笔记之Stream模块

2022-06-04 17:30

关注

一,开篇分析

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

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

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

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

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

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

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

var http = require('http') ;

  var server = http.createServer(function(req,res){

     res.writeHeader(200, {'Content-Type': 'text/plain'}) ;

     res.end("Hello,大熊!") ;

  }) ;

  server.listen(8888) ;

  console.log("http server running on port 8888 ...") ;

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

修改一下就好:

var http = require('http') ;

 var server = http.createServer(function(req,res){

    res.writeHeader(200,{

        'Content-Type' : 'text/plain;charset=utf-8'  // 添加charset=utf-8

    }) ;

    res.end("Hello,大熊!") ;

 }) ;

 server.listen(8888) ;

 console.log("http server running on port 8888 ...") ;

运行结果:

查看图片

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

var http = require('http') ;

 var fs = require('fs') ;

 var server = http.createServer(function (req, res) {

     fs.readFile(__dirname + '/data.txt', function (err, data) {

         res.end(data);

     }) ;

 }) ;

 server.listen(8888) ;

代码可以实现需要的功能,但是服务在发送文件数据之前需要缓存整个文件数据到内存,如果"data.txt"文件很
大并且并发量很大的话,会浪费很多内存。因为用户需要等到整个文件缓存到内存才能接受的文件数据,这样导致
用户体验相当不好。不过还好(req,res)两个参数都是Stream,这样我们可以用fs.createReadStream()代替fs.readFile()。如下:

var http = require('http') ;

 var fs = require('fs') ;

 var server = http.createServer(function (req, res) {

     var stream = fs.createReadStream(__dirname + '/data.txt') ;

     stream.pipe(res) ;

 }) ;

 server.listen(8888) ;

.pipe()方法监听fs.createReadStream()的'data' 和'end'事件,这样"data.txt"文件就不需要缓存整
个文件,当客户端连接完成之后马上可以发送一个数据块到客户端。使用.pipe()另一个好处是可以解决当客户
端延迟非常大时导致的读写不平衡问题。

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

二,实例引入

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

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

var rs = fs.createReadStream(pathname);

 rs.on('data', function (chunk) {

     doSomething(chunk) ; // 具体细节自己任意发挥

 });

 rs.on('end', function () {

     cleanUp() ;

 }) ;

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

var rs = fs.createReadStream(src) ;

 rs.on('data', function (chunk) {

     rs.pause() ;

     doSomething(chunk, function () {

         rs.resume() ;

     }) ;

 }) ;

 rs.on('end', function () {

     cleanUp();

 })  ;

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

此外,我们也可以为数据目标创建一个只写数据流,如下:

var rs = fs.createReadStream(src) ;

 var ws = fs.createWriteStream(dst) ;

 rs.on('data', function (chunk) {

     ws.write(chunk);

 }) ;

 rs.on('end', function () {

     ws.end();

 }) ;

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

var rs = fs.createReadStream(src) ;

 var ws = fs.createWriteStream(dst) ;

 rs.on('data', function (chunk) {

     if (ws.write(chunk) === false) {

         rs.pause() ;

     }

 }) ;

 rs.on('end', function () {

     ws.end();

 });

 ws.on('drain', function () {

     rs.resume();

 }) ;

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

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

var fs = require('fs'),

  path = require('path'),

  out = process.stdout;

var filePath = '/bb/bigbear.mkv';

var readStream = fs.createReadStream(filePath);

var writeStream = fs.createWriteStream('file.mkv');

var stat = fs.statSync(filePath);

var totalSize = stat.size;

var passedLength = 0;

var lastSize = 0;

var startTime = Date.now();

readStream.on('data', function(chunk) {

  passedLength += chunk.length;

  if (writeStream.write(chunk) === false) {

    readStream.pause();

  }

});

readStream.on('end', function() {

  writeStream.end();

});

writeStream.on('drain', function() {

  readStream.resume();

});

setTimeout(function show() {

  var percent = Math.ceil((passedLength / totalSize) * 100);

  var size = Math.ceil(passedLength / 1000000);

  var diff = size - lastSize;

  lastSize = size;

  out.clearLine();

  out.cursorTo(0);

  out.write('已完成' + size + 'MB, ' + percent + '%, 速度:' + diff * 2 + 'MB/s');

  if (passedLength < totalSize) {

    setTimeout(show, 500);

  } else {

    var endTime = Date.now();

    console.log();

    console.log('共用时:' + (endTime - startTime) / 1000 + '秒。');

  }

}, 500);

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

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

三,总结一下

(1),理解Stream概念。

(2),熟练使用相关Stream的api

(3),注意细节的把控,比如:大文件的拷贝,采用的使用 “chunk data” 的形式进行分片处理。

(4),pipe的使用

(5),再次强调一个概念:一个TCP连接既是可读流,又是可写流,而Http连接则不同,一个http request对象是可读流,而http response对象则是可写流。

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-前端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯