nodejs-流(stream)操作基础

什么是流？

• 流是一种用来处理文件的字节传输手段
• 它不关心文件的整体内容，只关注是否从文件中读取到了数据，以及读到数据之后的处理

流的类型？

• Readable--> 可读流(例如 fs.createReadStream())
• Writable--> 可写流(例如 fs.createWriteStream())
• Duplex--> 可读可写流(双工流 例如 net.Socket)
• Transform--> 在读写过程中可以修改和变化数据的Duplex流(例如 zlib.createDeflate())

Reanable(可读流)

创建可读流

var rs = fs.createReadStream(path,{
    flags: 'r', // 打开文件要做的操作，默认为‘r’
    encoding: 'utf8', // 指定解析的字符编码格式，默认为null; 同理：rs.setEncoding('utf8')
    start: '3', // 开始读取的索引位置
    end: '9', // 结束读取的索引位置（包括结束位置）
    highWaterMark: '3', // 从底层资源读取数据并存储在内部缓冲区中的最大字节数，默认16k；
                        //注意：如果指定utf8编码highWaterMark要大于3个字节
});

open事件

// 打开要读取的文件后触发
rs.on('open', () => {
    console.log('打开文件')
})

data事件

// 流自动从底层读取数据
rs.on('data', chunk => {
    console.log(chunk);
    // 调用pause()方法暂停数据的读取
    rs.pause(); // 此时切换为暂停模式
})
setTimeout(() => {
    // 恢复数据的读取，切换回流动模式
    rs.resume();
}, 2000)

error事件

// 读取数据过程中出现错误触发
rs.on('error', err => {
    console.log(err);
})

end事件

// 数据读取完毕的时候触发
rs.on('end', () => {
    console.log('读取结束');
})

close事件

rs.on('close', () => {
    console.log('读取完毕后关闭文件');
})

readable事件

rs.on('readable', () => {
    // 监听readable事件切换为暂停模式，调用read()方法读取流中缓存的数据
    // 当缓存中的数据不够highWaterMark，重新向底层读取highWaterMark字节的数据填充缓存区；
    // 缓存区的字节数可能会大于highWaterMark
    var chunk = rs.read(1)
    console.log(chunk);
})

可读流分为两种模式：

• flowing(流动模式)：
  当可读流监听"data"事件的时候，当前流为流动模式；可读流自动从系统底层读取数据，并通过EventEmitter发送事件来将数据提供给应用。

• paused(暂停模式)：
  当可读流监听"readable"事件的时候，当前流为暂停模式；必须要调用stream.read()方法从流中读取数据片段。

readable._readableState.flowing 字段的值来标识当前可读流是什么模式

• null：值为null情况下，可读流将不会产生数据，因为数据不会被消费；在当前状态下，监听流的"data"事件即可变为流动模式
• true：值为true的情况下，可读流为流动模式，流会自动读取数据返回
• false：值为false的情况下，可读流为暂停模式，只有调用stream.read()方法才可以从流中读取数据

Writable(可写流)

创建可写流

var ws = fs.createWriteStream(path, {
    flags: 'w', // 打开文件要做的操作，默认是‘w’
    encoding: 'utf8', // 指定写入的字符编码格式
    highWaterMark: '3', // 缓存区大小(默认为16kb), 
});

write方法

* chunk 要写入的数据,类型为 buffer/string
* encoding 可选，chunk为字符串时，指定字符编码
* callback 写入完毕后的回调
var flag = ws.write(chunk, encoding, callback);
// flag 为布尔值，缓存区满时为false，否则为true

end方法

ws.end(chunk, encoding, callback);
// 结束写入的方法，在结束的时候还可以写一部分数据进去，
// callback 如果传入，它将作为finish事件的回调函数

drain事件

var flag = ws.write(chunk, encoding, callback);
// 当flag为false时，表示缓存区已满；当缓存区数据用完，缓存区清空的时候会触发drain事件
// 必须是在缓存区满了清空后才会触发drain事件
ws.on('drain', () => {
    console.log('缓存区已清空')
})

finish事件

ws.end('结束');
ws.on('finish', () => {
    console.log('所有写入完成');
})
// 在调用了 stream.end() 方法，且缓冲区数据都已经传给底层系统之后， 'finish' 事件将被触发。

pipe

Readable和Writable分别实现了对文件的读和写的操作；但是通常情况下会出现边读边写的场景，读取一个文件的内容，写入到另一个文件中；在这种场景下，可能会出现读写不均衡的问题，写入比较慢，读取比较快(来不及写入的文件数据可能会丢失)；所以我们期待可以达到读写均衡的状态，于是出现了pipe(导流)。
它可以控制读取的速率，当写入较慢的时候暂停对文件的读取；当可写流缓存区数据写入完毕后恢复文件的读取。

pipe的用法

readStream.pipe(writeStream);
var from = fs.createReadStream('./1.txt');
var to = fs.createWriteStream('./2.txt');
from.pipe(to);
// pipe可以绑定多个可写流
var to2 = fs.createWriteStream('./3.txt');
from.pipe(to2);
// 分离from绑定的可写流;
// 不传参数的话会分离所有绑定的可写流
from.unpipe(to)

pipe原理

// 当可写流调用write()方法返回false时，表示缓存区已满，这时将可读流切换为暂停模式;
// 暂停读取数据，同时监听可写流的drain事件，当缓存区数据写入完毕，触发drain事件；
// 在drain事件的回调函数中切换可读流为流动模式继续读取数据
var fs = require('fs');
var ws = fs.createWriteStream('./2.txt');
var rs = fs.createReadStream('./1.txt');
rs.on('data', data => {
    var flag = ws.write(data);
    if(!flag)
    rs.pause();
});
ws.on('drain', () => {
    rs.resume();
});
rs.on('end', () => {
    ws.end();
});

Duplex(双工流)

Duplex 流是同时实现了 Readable 和 Writable 接口的流
其中Readable和Writable分别是两个不相关的流

// 实现一个简单自定义的duplex需要定义好两个方法read和write

let {Duplex} = require('stream');
let index = 0;
let s = Duplex({
    read(){
        console.log(index)
        if(index++<3){
            this.push('b');
        } else {
            this.push(null); 
        }
    },
    write(chunk,encoding,cb){
        var a = chunk.toString().toUpperCase()
        console.log(1);
        cb();
    }
});
//process.stdin 标准输入流
//proces.stdout标准输出流
process.stdin.pipe(s)
s.pipe(process.stdout)
process.stdin.pipe(s).pipe(process.stdout);

Transform(转换)

transform流也是一个双工流，用以处理输入输出是因果相关，位于管道中间层的 Transform 是即可读也可写的；
Transform类最初继承自stream.Duplex，并且实现了它自己版本的writable._write()和readable._read()方法。
自定义一个transform流必须实现transform() 方法；

let {Transform}  = require('stream');
//转换流是实现数据转换的
let t = Transform({
    transform(chunk,encoding,cb){
        this.push(chunk.toString().toUpperCase());
        cb();
    }
});
process.stdin.pipe(t).pipe(process.stdout);

Transform 同样是双工流，看起来和 Duplex 重复了，但两者有一个重要的区别：Duplex 虽然同时具备可读流和可写流，但两者是相对独立的；Transform 的一种流的数据会经过一定的处理过程自动进入另外一个流。

transform.png

参考：

深入理解 Node Stream 内部机制
nodejs中流(stream)的理解
基础
进阶
实战