NodeJS 学习笔记之 stream

发布于 2021-11-17 12:51:17 字数 9095 浏览 1166 评论 0

本章节学习流，流的一个好处在于减少各种异步IO的回调地狱。IO操作遍及我们各种操作，比如数据库读写，文件读写，文件转换压缩……别的不说，比如第一节，我们要将一个HTML文件返回浏览器，就涉及IO操作。

场景

一个页面，如果按版本划分功能，可能切成一块块给不同的人做，使用fs方法的异步IO方法，可能是这样写的：

fs.asyncXXX(function(err,data){
   fs.asyncXXX(function(err,data){
      fs.asyncXXX(function(err,data){
         fs.asyncXXX(function(err,data){
 
         })
      })
   })
})

如果使用流，则是这样写

readStream.pipe(transformStream).pipe(writeStream)

a.pipe(b).pipe(c).pipe(d)
//相当于
a.pipe(b);
b.pipe(c);
c.pipe(d);

这是不是与 Unix 中的管道很相似呢？！无错，它的灵感就在于这！

a | b | c | d

此外，不使用流，如果读出一个很大的文件，则需要将它整个装进内存中，这会很影响性能。使用了流就不用担心这个。

概念

nodejs底层一个提供了4个流， Readable 流、Writable 流、Duplex 流和 Transform 流。如果还不能满足你的需求，可以自己继承其个流的基类进行扩展。（例如util.inherits(MyTransform, Transform); ）

使用情景	类	需要重写的方法
只读	`Readable`	`_read`
只写	`Writable`	`_write`
双工	`Duplex`	`_read`, `_write`
操作被写入数据，然后读出结果	`Transform`	`_transform`, `_flush`

说了这么多，我们还没有瞧见一个流。我们来一些例子吧.

可读流

//aaa.js
var fs = require('fs');
var readStream = fs.createReadStream('myfile.txt');//里面乱写几行
readStream.pipe(process.stdout);

上面是直接读文本的，然后输出到控制台。我们也可以加密一下输出

var crypto = require('crypto');
var fs = require('fs');
 
var readStream = fs.createReadStream('myfile.txt');
var hash = crypto.createHash('sha1');
readStream
  .on('data', function (chunk) {
    hash.update(chunk);
  })
  .on('end', function () {
    console.log(hash.digest('hex'));
  });

输出一大堆看不懂的密码。fs模块也有创建可读流的方法：

var fs = require('fs');
var readableStream = fs.createReadStream('file.txt');
var data = '';
 
readableStream.setEncoding('utf8');
 
readableStream.on('data', function(chunk) {
    data+=chunk;
});
 
readableStream.on('end', function() {
    console.log(data);
});

我们再看一下可读流的各种事件

var fs = require('fs');
 
var readStream = fs.createReadStream('myfile.txt');
readStream
  .on('data', function (chunk) {
    console.log("emit data")
    console.log(chunk.toString('utf8'))
  })
  .on('end', function () {
    console.log("emit end");
  })
  .on('close', function () {
    console.log("emit close");
  })
  .on("readable", function(){
     console.log("emit readable")
  })
  .on("error", function(e){
     console.log("emit error")
  })

再看一下如何重写 _read 方法：

var Readable = require('stream').Readable;
var util = require('util');
 
function CountingObjectStream(length, options) {
  if (!(this instanceof CountingObjectStream)) {
    return new CountingObjectStream(length, options);
  }
  if (!options) options = {}; // ensure object
  options.objectMode = true; // forcing object mode
  Readable.call(this, options);
  this.lenToCount = length;  // how far to count
  this.index = 0;  // to track our count
}
util.inherits(CountingObjectStream, Readable);
 
CountingObjectStream.prototype._read = function () {
  this.index += 1;
  if (this.index > this.lenToCount) {
    return this.push(null); // done, return
  }
 
  // pushing number, but could be any non-null obj
  this.push(this.index);
};
 
 
// consume this stream and output to stdout
// coercing it to a string
var readStream = new CountingObjectStream(10);
readStream
  .on('readable', function () {
    var obj;
    while (null !== (obj = readStream.read())) {
      console.log(obj);
    }
  });

Readable 有一个可选的 hash 参数里，里面有三个配置项：

highWaterMark {Number} 停止从底层资源读取前内部缓冲区最多能存放的字节数。缺省为 16kb，对于 objectMode 流则是 16
-encoding {String} 若给出，则 Buffer 会被解码成所给编码的字符串。缺省为 null
objectMode {Boolean} 该流是否应该表现为对象的流。意思是说 stream.read(n) 返回一个单独的对象，而不是大小为 n 的 Buffer

前两个配置项比较易懂，我们看第三个：

var stream = require('stream');
var util = require('util');
 
function StringifyStream(){
    stream.Transform.call(this);
 
    this._readableState.objectMode = false;
    this._writableState.objectMode = true;
}
util.inherits(StringifyStream, stream.Transform);
 
StringifyStream.prototype._transform = function(obj, encoding, cb){
    this.push(JSON.stringify(obj));
    cb();
};

 
var json = require(__dirname + 'test.json');
console.log(json) //这是一个对象
var rs = new stream.Readable({ objectMode: true });
rs.push(json);
rs.push(null);
 
rs.pipe(new StringifyStream()).pipe(process.stdout);

下面是 test.json

{
"a":"2",
"b":{
      "xxx": 1,
       "yyy": false
   }
}

可写流

构造器有一个可选配置对象，默认是编码是 utf8

var fs = require('fs');
var wstream = fs.createWriteStream('myOutput.txt');
wstream.write('Hello world!\n');
wstream.write('Another line\n');
wstream.end();

我们可以这样改编码

var fs = require('fs');
var wstream = fs.createWriteStream('myOutput.txt');
wstream.write('Hello world!\n');
wstream.write('Another line\n');
wstream.end();

输出二进制文件

var crypto = require('crypto');
var fs = require('fs');
var wstream = fs.createWriteStream('myBinaryFile');
// creates random Buffer of 100 bytes
var buffer = crypto.randomBytes(100);
wstream.write(buffer);
// create another Buffer of 100 bytes and write
wstream.write(crypto.randomBytes(100));
wstream.end();